fix(ingestion): contain process-fatal errors and bound parser work (#21480)

internal/agentic_rag/decisionhead/session.go

@@ -204,16 +204,24 @@ func (s *Session) run(ctx context.Context, seq Sequence) ([]float32, error) {
 	}
 	defer runOpts.Destroy()
 	// Terminate the ORT run when ctx is done; done closes once run returns so
-	// the watcher exits on the success path too.
+	// the watcher exits on the success path too. Join the watcher before
+	// runOpts.Destroy() executes (defers run LIFO) so Terminate() can never
+	// touch a freed handle.
 	done := make(chan struct{})
-	defer close(done)
+	var watcher sync.WaitGroup
+	watcher.Add(1)
 	go func() {
+		defer watcher.Done()
 		select {
 		case <-ctx.Done():
 			_ = runOpts.Terminate()
 		case <-done:
 		}
 	}()
+	defer func() {
+		close(done)
+		watcher.Wait()
+	}()
 
 	outputs := []ort.Value{nil, nil}
 	inputs := []ort.Value{inIDs, inAttention, inPos, inMarkerMask, inQType}

internal/deepdoc/native/ocr_rec.go

@@ -341,17 +341,9 @@ func (s *recSession) Run(ctx context.Context, input []float32) ([]float32, error
 		return nil, err
 	}
 	defer opts.Destroy()
-	// Cancel an in-flight Run when the context is done. done closes once Run
-	// returns so the watcher exits even on the success path.
-	done := make(chan struct{})
-	defer close(done)
-	go func() {
-		select {
-		case <-ctx.Done():
-			_ = opts.Terminate()
-		case <-done:
-		}
-	}()
+	// Terminate an in-flight Run on cancellation, and join the watcher before
+	// opts.Destroy() runs (defers are LIFO).
+	defer watchRunOptions(ctx, opts)()
 
 	// Fresh input tensor per Run; freed right after the call. nil output →
 	// onnxruntime allocates the actual-shaped tensor.

internal/deepdoc/native/run_options.go

@@ -0,0 +1,32 @@
+//go:build cgo
+
+package native
+
+import (
+	"context"
+	"sync"
+
+	ort "github.com/infiniflow/onnxruntime_go"
+)
+
+// watchRunOptions terminates opts when ctx is cancelled and returns a stop
+// function that joins the watcher. Callers must defer the returned function
+// after the deferred opts.Destroy(): defers run LIFO, so the join executes
+// first and Terminate() can never touch a freed RunOptions handle.
+func watchRunOptions(ctx context.Context, opts *ort.RunOptions) func() {
+	done := make(chan struct{})
+	var watcher sync.WaitGroup
+	watcher.Add(1)
+	go func() {
+		defer watcher.Done()
+		select {
+		case <-ctx.Done():
+			_ = opts.Terminate()
+		case <-done:
+		}
+	}()
+	return func() {
+		close(done)
+		watcher.Wait()
+	}
+}

internal/deepdoc/native/session.go

@@ -304,17 +304,9 @@ func (s *session) Run(ctx context.Context, input []float32) ([]float32, error) {
 		return nil, err
 	}
 	defer opts.Destroy()
-	// Cancel an in-flight Run when the context is done. done closes once Run
-	// returns so the watcher exits even on the success path.
-	done := make(chan struct{})
-	defer close(done)
-	go func() {
-		select {
-		case <-ctx.Done():
-			_ = opts.Terminate()
-		case <-done:
-		}
-	}()
+	// Terminate an in-flight Run on cancellation, and join the watcher before
+	// opts.Destroy() runs (defers are LIFO).
+	defer watchRunOptions(ctx, opts)()
 
 	// Allocate a fresh input tensor for this Run and free it right after the
 	// call. The output is passed as nil so ORT allocates it; we copy the data

internal/deepdoc/parser/pdf/layout/combined_column.go

@@ -125,12 +125,28 @@ const maxPageExtent = 14400.0
 // as-is: the "anomalies" are the norm, not noise.
 const maxTrimFraction = 0.10
 
-// maxBins caps the histogram allocation in gapColumnCount/detectColumnCount2D so
-// a malformed (ballooned) page extent cannot trigger an OOM-scale allocation
-// (mirrors pdf-inspector's bin cap). When the extent is huge, the bin is
-// widened so the projection still resolves real gutters.
+// maxBins caps the histogram allocation in gapColumnWidths/gapColumnCount/
+// detectColumnCount2D so a malformed (ballooned) page extent cannot trigger an
+// OOM-scale allocation (mirrors pdf-inspector's bin cap). When the extent is
+// huge, the bin is widened so the projection still resolves real gutters.
 const maxBins = 65536
 
+// columnHistogramBins returns the bin width and bin count for a page extent,
+// widening the bin when width/binPt would exceed maxBins. All three column
+// projections (gapColumnWidths, gapColumnCount, detectColumnCount2D) share this
+// so the allocation bound and the bin geometry cannot drift apart.
+func columnHistogramBins(width, baseBin float64) (float64, int) {
+	effBin := baseBin
+	if width/float64(maxBins) > effBin {
+		effBin = width / float64(maxBins)
+	}
+	nb := int(width/effBin) + 1
+	if nb < 1 {
+		nb = 1
+	}
+	return effBin, nb
+}
+
 // gapMinFrac: a horizontal run of low coverage counts as a column gap only if
 // it is at least this fraction of the page width. Reused by both gapColumnCount
 // and the 2D both-sides gutter rescue so the two detectors agree on what a
@@ -152,7 +168,10 @@ func detectColumnCount(boxes []pdf.TextBox, indices []int) (int, []float64) {
 	}
 	g := gapColumnCount(lines, gapMinFrac, 0.15, binPt)
 	if g >= 2 {
-		_, width := pageExtent(lines)
+		// Use the same robust text extent as the gap/width projections: a
+		// malformed box must not inflate the denominator and make real columns
+		// look like narrow table cells.
+		_, width := robustPageExtent(textProjectionLines(lines))
 		if width > 0 {
 			widths := gapColumnWidths(lines)
 			maxw := 0.0
@@ -261,16 +280,27 @@ func gapColumnWidths(lines []pdf.TextBox) []float64 {
 	if n == 0 {
 		return nil
 	}
-	minX0, width := pageExtent(lines)
+	// Mirror gapColumnCount's projection: drop figure/equation placeholders and
+	// derive the extent from the robust text body. Using the raw extent here let
+	// one malformed box set the bin geometry, so ordinary lines collapsed into
+	// the first bin and the width gate disagreed with the gap count.
+	lines = textProjectionLines(lines)
+	if len(lines) == 0 {
+		return nil
+	}
+	// Threshold must be relative to the projected text lines, not the pre-filter
+	// count: figure/equation placeholders can otherwise push thr above the
+	// maximum per-column coverage and yield no widths.
+	n = len(lines)
+	minX0, width := robustPageExtent(lines)
 	if width <= 0 {
 		return nil
 	}
-	binPt := 2.0
-	nb := int(width/binPt) + 1
+	effBin, nb := columnHistogramBins(width, binPt)
 	cov := make([]int, nb)
 	for _, b := range lines {
-		i0 := clampInt(int((b.X0-minX0)/binPt), 0, nb-1)
-		i1 := clampInt(int((b.X1-minX0)/binPt), 0, nb-1)
+		i0 := clampInt(int((b.X0-minX0)/effBin), 0, nb-1)
+		i1 := clampInt(int((b.X1-minX0)/effBin), 0, nb-1)
 		for i := i0; i <= i1; i++ {
 			cov[i]++
 		}
@@ -287,7 +317,9 @@ func gapColumnWidths(lines []pdf.TextBox) []float64 {
 		for j < nb && float64(cov[j]) >= thr {
 			j++
 		}
-		widths = append(widths, float64(j-i)*binPt)
+		// Bins were widened by columnHistogramBins; convert bin counts with the
+		// effective width, not the base resolution.
+		widths = append(widths, float64(j-i)*effBin)
 		i = j
 	}
 	return widths
@@ -319,6 +351,10 @@ func gapColumnCount(lines []pdf.TextBox, gapMinFrac, crossTol, binPt float64) in
 	if len(lines) == 0 {
 		return 1
 	}
+	// Threshold must be relative to the projected text lines, not the pre-filter
+	// count: figure/equation placeholders can otherwise push thr above the
+	// maximum per-column coverage and make the two detectors disagree.
+	n = len(lines)
 	// A2: robust page extent discards malformed/outlier lines so a single bad
 	// box cannot balloon the width and collapse detection to one column.
 	minX0, width := robustPageExtent(lines)
@@ -326,17 +362,7 @@ func gapColumnCount(lines []pdf.TextBox, gapMinFrac, crossTol, binPt float64) in
 		return 1
 	}
 	minGap := gapMinFrac * width
-	// Cap the bin count so a ballooned extent cannot allocate an OOM-scale
-	// histogram. When the extent is huge, widen the bin so the projection still
-	// resolves real gutters.
-	effBin := binPt
-	if width/float64(maxBins) > effBin {
-		effBin = width / float64(maxBins)
-	}
-	nb := int(width/effBin) + 1
-	if nb < 1 {
-		nb = 1
-	}
+	effBin, nb := columnHistogramBins(width, binPt)
 	cov := make([]int, nb)
 	for _, b := range lines {
 		i0 := int((b.X0 - minX0) / effBin)
@@ -440,11 +466,7 @@ func detectColumnCount2D(lines []pdf.TextBox) (int, []float64) {
 	// Cap the bin count so a ballooned extent cannot allocate an OOM-scale
 	// histogram. When the extent is huge, widen the bin so the projection still
 	// resolves real gutters.
-	effBin := binPt
-	if width/float64(maxBins) > effBin {
-		effBin = width / float64(maxBins)
-	}
-	nb := int(width/effBin) + 1
+	effBin, nb := columnHistogramBins(width, binPt)
 	proj := make([]int, nb)
 	for _, b := range body {
 		w := utf8.RuneCountInString(b.Text)

internal/deepdoc/parser/pdf/layout/combined_column_defensive_test.go

@@ -1,6 +1,7 @@
 package layout
 
 import (
+	"math"
 	"testing"
 
 	pdf "ragflow/internal/deepdoc/parser/pdf/type"
@@ -28,6 +29,126 @@ func addBox(boxes *[]pdf.TextBox, x0, x1 float64) {
 	})
 }
 
+// TestGapColumnWidthsCapsHistogram proves the OOM guard: an absurd page extent
+// must not allocate one histogram bin per point. The call is expected to return
+// quickly with a bounded bin count instead of trying a multi-GB allocation.
+func TestGapColumnWidthsCapsHistogram(t *testing.T) {
+	// The shared bin helper must widen the bin instead of allocating one bin
+	// per point for an absurd extent.
+	effBin, nb := columnHistogramBins(1e9, binPt)
+	if nb > maxBins+1 {
+		t.Fatalf("bin count %d exceeds cap %d", nb, maxBins)
+	}
+	if effBin <= binPt {
+		t.Fatalf("bin width %v was not widened for a 1e9 extent", effBin)
+	}
+	// Normal page extents keep the configured resolution.
+	if eff, n := columnHistogramBins(600, binPt); eff != binPt || n != 301 {
+		t.Fatalf("normal extent bins = (%v, %d), want (%v, 301)", eff, n, binPt)
+	}
+
+	var lines []pdf.TextBox
+	for i := 0; i < 20; i++ {
+		lines = append(lines, pdf.TextBox{X0: 100, X1: 300})
+	}
+	for i := 0; i < 20; i++ {
+		lines = append(lines, pdf.TextBox{X0: 400, X1: 600})
+	}
+	lines = append(lines, pdf.TextBox{X0: 100, X1: 1e9}) // malformed extent
+
+	widths := gapColumnWidths(lines)
+	if len(widths) != 2 {
+		t.Fatalf("gapColumnWidths returned %d widths for a two-column page, want 2 (%v)", len(widths), widths)
+	}
+	for _, w := range widths {
+		if math.Abs(w-200) > 10 {
+			t.Fatalf("column width %v, want ~200 (outlier must not distort the projection)", w)
+		}
+	}
+
+	// A huge but internally consistent extent exercises the widened-bin path:
+	// two 1000pt columns 1,000,000pt apart. Widths must be reported in page
+	// units (effBin), not in base 2pt bins.
+	lines = nil
+	for i := 0; i < 20; i++ {
+		lines = append(lines, pdf.TextBox{X0: 0, X1: 1000})
+	}
+	for i := 0; i < 20; i++ {
+		lines = append(lines, pdf.TextBox{X0: 999000, X1: 1000000})
+	}
+	widths = gapColumnWidths(lines)
+	if len(widths) != 2 {
+		t.Fatalf("widened-bin page returned %d widths, want 2 (%v)", len(widths), widths)
+	}
+	for _, w := range widths {
+		if math.Abs(w-1000) > 50 {
+			t.Fatalf("widened-bin column width %v, want ~1000", w)
+		}
+	}
+}
+
+// TestDetectColumnCount_OutlierKeepsWidthGate proves the width gate uses the
+// robust text extent: an outlier that gapColumnCount ignores must not inflate
+// the denominator and collapse a real two-column page to one column.
+func TestDetectColumnCount_OutlierKeepsWidthGate(t *testing.T) {
+	var boxes []pdf.TextBox
+	for i := 0; i < 20; i++ {
+		addBox(&boxes, 100, 300)
+	}
+	for i := 0; i < 20; i++ {
+		addBox(&boxes, 400, 600)
+	}
+	boxes = append(boxes, pdf.TextBox{
+		PageNumber: 0, X0: 100, X1: 1e9, Top: 9999, Bottom: 10009,
+		LayoutType: pdf.LayoutTypeText, Text: "bad",
+	})
+	indices := make([]int, len(boxes))
+	for i := range indices {
+		indices[i] = i
+	}
+	if k, _ := detectColumnCount(boxes, indices); k != 2 {
+		t.Fatalf("detectColumnCount = %d, want 2 (outlier must not inflate the width gate)", k)
+	}
+}
+
+// TestColumnProjectionsIgnoreFigureCountInThreshold proves the coverage
+// threshold is computed from projected text lines: figure placeholders must not
+// raise thr above the per-column coverage and starve the width projection.
+func TestColumnProjectionsIgnoreFigureCountInThreshold(t *testing.T) {
+	var boxes []pdf.TextBox
+	for i := 0; i < 20; i++ {
+		addBox(&boxes, 100, 300)
+	}
+	for i := 0; i < 20; i++ {
+		addBox(&boxes, 400, 600)
+	}
+	for i := 0; i < 100; i++ {
+		top := float64(len(boxes)) * 12
+		boxes = append(boxes, pdf.TextBox{
+			PageNumber: 0, X0: 100, X1: 600, Top: top, Bottom: top + 10,
+			LayoutType: pdf.LayoutTypeFigure, Text: "fig",
+		})
+	}
+
+	widths := gapColumnWidths(boxes)
+	if len(widths) != 2 {
+		t.Fatalf("gapColumnWidths returned %d widths, want 2 (%v)", len(widths), widths)
+	}
+	for _, w := range widths {
+		if math.Abs(w-200) > 10 {
+			t.Fatalf("column width %v, want ~200", w)
+		}
+	}
+
+	indices := make([]int, len(boxes))
+	for i := range indices {
+		indices[i] = i
+	}
+	if k, _ := detectColumnCount(boxes, indices); k != 2 {
+		t.Fatalf("detectColumnCount = %d, want 2 (figures must not raise the threshold)", k)
+	}
+}
+
 // TestTextProjectionLinesStripsFigures proves A1: image and equation
 // placeholders must not feed the column projection (a figure spanning the
 // gutter would otherwise mask a real column boundary). Text and table boxes

internal/deepdoc/parser/pdf/pdfium/pdfium.go

@@ -140,6 +140,15 @@ func RenderPage(pdfData []byte, pageIdx int, dpi float64) (out *image.RGBA, err
 		pxW := int(math.Round(pw * scale))
 		pxH := int(math.Round(ph * scale))
 
+		// Reject absurd pages before pdfium allocates the bitmap. RenderPage
+		// holds two buffers at peak: pdfium's pxH*stride bitmap and the
+		// image.NewRGBA copy (4 bytes/pixel each), so budget 8 bytes/pixel.
+		const maxRenderBytes = 1 << 30
+		if int64(pxW)*int64(pxH)*8 > maxRenderBytes {
+			err = fmt.Errorf("pdfium: page %d render buffers %dx%d exceed %d bytes", pageIdx, pxW, pxH, maxRenderBytes)
+			return
+		}
+
 		bitmap := C.FPDFBitmap_Create(C.int(pxW), C.int(pxH), 1) // 1 = RGBA
 		if bitmap == nil {
 			err = fmt.Errorf("pdfium: FPDFBitmap_Create(%d,%d) returned nil", pxW, pxH)
@@ -151,7 +160,13 @@ func RenderPage(pdfData []byte, pageIdx int, dpi float64) (out *image.RGBA, err
 		// (e.g. outside crop box) are white rather than undefined.
 		stride := int(C.FPDFBitmap_GetStride(bitmap))
 		buf := C.FPDFBitmap_GetBuffer(bitmap)
-		pixels := (*[1 << 30]byte)(unsafe.Pointer(buf))[: pxH*stride : pxH*stride]
+		if stride <= 0 || buf == nil {
+			err = fmt.Errorf("pdfium: bitmap %dx%d has invalid buffer (stride %d)", pxW, pxH, stride)
+			return
+		}
+		// pdfium owns a pxH*stride byte buffer. Describe it directly instead of
+		// slicing a fixed [1<<30]byte array, which panicked on large pages.
+		pixels := unsafe.Slice((*byte)(buf), pxH*stride)
 		for i := range pixels {
 			pixels[i] = 255
 		}

internal/ingestion/knowledge_compile/consumer.go

@@ -778,16 +778,14 @@ func (c *Consumer) processBatch(ctx context.Context, tenant, kb, token string, e
 		// A no-op deduper would silently disable dataset-level LLM merging: every
 		// candidate (including e.g. a "吕布" wiki_page) would be written as its own
 		// merged row and duplicates would accumulate across runs. That degradation
-		// is never acceptable here, so fail loudly instead of papering over it.
-		common.Fatal("knowledge_compile: dataset-level LLM deduper unavailable, refusing to continue with a no-op merge",
-			zap.String("kb_id", kb),
-			zap.String("tenant_id", tenant),
-			zap.String("factory_err", func() string {
-				if err != nil {
-					return err.Error()
-				}
-				return "deduper factory returned nil"
-			}()))
+		// is never acceptable here, so fail the batch instead of papering over it.
+		// Never call common.Fatal here: the ingestor is shared by every tenant, so
+		// one tenant's model misconfiguration must not terminate the process.
+		factoryErr := "deduper factory returned nil"
+		if err != nil {
+			factoryErr = err.Error()
+		}
+		return fmt.Errorf("knowledge_compile: dataset-level LLM deduper unavailable for kb %s (tenant %s): %s", kb, tenant, factoryErr)
 	}
 
 	// --- Completion merge ---

internal/ingestion/knowledge_compile/service.go

@@ -129,9 +129,9 @@ func InitializePublisher(db *gorm.DB, mq engine.MessageQueue) {
 }
 
 // defaultDeduperFactory resolves the per-tenant LLM deps and builds the
-// KB-scoped deduper. On any failure it returns an error so the caller falls
-// back to the noop deduper (merged products are still written, just without
-// cross-document LLM merging).
+// KB-scoped deduper. On any failure it returns an error so the caller can fail
+// the batch: a noop deduper would silently disable cross-document LLM merging
+// and let duplicate merged products accumulate.
 func defaultDeduperFactory(ctx context.Context, tenant string) (Deduper, error) {
 	deps, err := kccommon.ResolveDeps(ctx, tenant, defaultLLMID, defaultEmbedding)
 	if err != nil {

internal/utility/scheduled_task.go

@@ -17,6 +17,7 @@
 package utility
 
 import (
+	"fmt"
 	"ragflow/internal/common"
 	"sync"
 	"sync/atomic"
@@ -90,7 +91,10 @@ func (t *ScheduledTask) runSafely() {
 
 	defer func() {
 		if r := recover(); r != nil {
-			common.Fatal("Task panicked", zap.String("name", t.Name), zap.Any("recover", r))
+			// A periodic job must never take the whole service down: log the panic
+			// and let the next tick run. common.Fatal would call os.Exit(1), which
+			// turns one bad tick into an outage for every tenant on this process.
+			common.Error("Task panicked", fmt.Errorf("panic: %v", r), zap.String("name", t.Name), zap.Any("recover", r))
 		}
 	}()
 

internal/utility/scheduled_task_test.go

@@ -23,6 +23,30 @@ import (
 	"time"
 )
 
+func TestScheduledTaskPanicIsContained(t *testing.T) {
+	var runs atomic.Int32
+	task := NewScheduledTask("panic", time.Hour, func() {
+		runs.Add(1)
+		panic("boom")
+	})
+	t.Cleanup(task.Stop)
+
+	// A panicking job must be contained (logged), not escalated to os.Exit.
+	task.runSafely()
+	if got := runs.Load(); got != 1 {
+		t.Fatalf("job ran %d times, want 1", got)
+	}
+	if task.IsExecuting() {
+		t.Fatal("executing flag was not reset after a panic")
+	}
+
+	// The task must remain usable after the panic.
+	task.runSafely()
+	if got := runs.Load(); got != 2 {
+		t.Fatalf("job ran %d times after a panic, want 2", got)
+	}
+}
+
 func TestScheduledTaskConcurrentStart(t *testing.T) {
 	task := NewScheduledTask("concurrent start", time.Hour, func() {})
 	t.Cleanup(task.Stop)