Skip to content

Commit 9f9e087

Browse files
committed
fix(ingest): per-LLM-call timeout + leaf-section cap (un-stick big PDFs)
Two ingest bugs that froze FinanceBench ingests and are real product defects on any large filing: 1. No per-LLM-call timeout. A single hung summarize / HyDE / multi-axis / TOC-build call blocked the stage's errgroup Wait() forever — a doc was observed stuck in `summarizing` for 13+ hours. Fix: completeWithTimeout wraps every individual LLM.Complete in a context.WithTimeout (default 90s, ingest.llm_call_timeout_seconds / VLE_INGEST_LLM_CALL_TIMEOUT_SECONDS). On timeout the call is logged and skipped — the section keeps its existing/empty summary and the document still reaches `ready`. One bad call can no longer freeze a whole document. 2. Leaf-section explosion. chunkOversizedLeaves splits any leaf over 2400 chars into ~900-char pieces, so a 45K-char "Notes to Financial Statements" section shattered into ~50 chunks; a 92-page 10-K produced ~1500 leaves, each costing a summarize+HyDE+multi-axis LLM call → the slow/stalled ingest. Fix: capLeafSections enforces a ceiling (default 400, ingest.max_sections / VLE_INGEST_MAX_SECTIONS) by merging the smallest adjacent leaf siblings under a shared parent until the count is within budget. Content is preserved (blank-line joined), page ranges unioned, and table sections — attached after this pass — are never merged. Applied in both the heuristic and outline parse paths. The cap runs at its default (400) through the existing RegistryFromTableOpts → NewPDFWithTables path, so the fix is active on the deployed binary without a cmd wiring change. ingest.max_sections becoming operator-tunable end-to-end is a small follow-up in the cmd binaries. Tests: a hung-call mock proves the pipeline still completes and other sections summarize; cap tests prove merge-down to budget, smallest-pair ordering, content preservation, and the disabled (<=0) escape hatch. go build/vet/test all green.
1 parent e183ca7 commit 9f9e087

8 files changed

Lines changed: 414 additions & 31 deletions

File tree

pkg/config/config.go

Lines changed: 50 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,37 @@ type IngestConfig struct {
6565
// 4 + 4 per-stage caps while staying well below typical provider
6666
// per-tenant concurrency limits.
6767
GlobalLLMConcurrency int `yaml:"global_llm_concurrency"`
68+
69+
// LLMCallTimeoutSeconds bounds each INDIVIDUAL LLM call issued by the
70+
// ingest pipeline (one section's summary, one leaf's HyDE questions,
71+
// one TOC detect/extract/verify turn). Without it, a single provider
72+
// call that hangs — no response, no error — blocks its bounded-
73+
// concurrency errgroup forever, and the document never leaves
74+
// `summarizing`. (Observed: a doc stuck for 13+ hours.)
75+
//
76+
// When a call exceeds this deadline it is treated exactly like any
77+
// other per-section failure: logged and skipped, leaving that
78+
// section with its existing/empty summary. One bad section can no
79+
// longer freeze the whole document — it still reaches `ready`.
80+
//
81+
// 0 (or omitted) defaults to 90. Set explicitly to tune for slow
82+
// reasoning models; a negative value is rejected by Validate.
83+
LLMCallTimeoutSeconds int `yaml:"llm_call_timeout_seconds"`
84+
85+
// MaxSections caps the number of leaf sections a single document may
86+
// produce. A pathological PDF (e.g. a 92-page 10-Q whose every bold
87+
// table cell looks like a heading) can shatter into ~1500 leaves,
88+
// each of which then costs a summarize + HyDE LLM call — thousands of
89+
// calls that throttle or stall ingest.
90+
//
91+
// When the parsed leaf count exceeds this cap, the parser merges
92+
// adjacent small leaf sections under a shared parent (smallest
93+
// siblings first) until the document is back under the cap.
94+
//
95+
// 0 (or omitted) defaults to 400 — comfortably above a real filing's
96+
// section count (~170-510 with tables) while still catching the
97+
// runaway case. A negative value is rejected by Validate.
98+
MaxSections int `yaml:"max_sections"`
6899
}
69100

70101
// TOCBlock configures the LLM-driven table-of-contents tree
@@ -658,7 +689,9 @@ func Default() Config {
658689
},
659690
},
660691
Ingest: IngestConfig{
661-
GlobalLLMConcurrency: 12,
692+
GlobalLLMConcurrency: 12,
693+
LLMCallTimeoutSeconds: 90,
694+
MaxSections: 400,
662695
HyDE: HyDEConfig{
663696
Enabled: true,
664697
NumQuestions: 5,
@@ -814,6 +847,16 @@ func applyEnvOverrides(c *Config) {
814847
c.Ingest.GlobalLLMConcurrency = n
815848
}
816849
}
850+
if v := os.Getenv("VLE_INGEST_LLM_CALL_TIMEOUT_SECONDS"); v != "" {
851+
if n, err := strconv.Atoi(v); err == nil && n >= 0 {
852+
c.Ingest.LLMCallTimeoutSeconds = n
853+
}
854+
}
855+
if v := os.Getenv("VLE_INGEST_MAX_SECTIONS"); v != "" {
856+
if n, err := strconv.Atoi(v); err == nil && n >= 0 {
857+
c.Ingest.MaxSections = n
858+
}
859+
}
817860
// pdftable-driven table extraction.
818861
if v := os.Getenv("VLE_INGEST_TABLES_ENABLED"); v != "" {
819862
switch strings.ToLower(strings.TrimSpace(v)) {
@@ -1091,6 +1134,12 @@ func (c Config) Validate() error {
10911134
if c.Ingest.GlobalLLMConcurrency < 0 {
10921135
return fmt.Errorf("ingest.global_llm_concurrency must be >= 0, got %d", c.Ingest.GlobalLLMConcurrency)
10931136
}
1137+
if c.Ingest.LLMCallTimeoutSeconds < 0 {
1138+
return fmt.Errorf("ingest.llm_call_timeout_seconds must be >= 0, got %d", c.Ingest.LLMCallTimeoutSeconds)
1139+
}
1140+
if c.Ingest.MaxSections < 0 {
1141+
return fmt.Errorf("ingest.max_sections must be >= 0, got %d", c.Ingest.MaxSections)
1142+
}
10941143

10951144
switch c.Ingest.Tables.VerticalStrategy {
10961145
case "", "lines", "lines_strict", "text", "explicit":

pkg/ingest/hyde.go

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import (
88
"io"
99
"strings"
1010
"sync"
11+
"time"
1112

1213
"golang.org/x/sync/errgroup"
1314

@@ -158,7 +159,7 @@ func (p *Pipeline) candidateQuestionsFor(ctx context.Context, s db.Section, prof
158159
JSONSchema: []byte(hydeJSONSchema),
159160
}
160161

161-
questions, err := runHyDEWithRetry(ctx, p.LLM, req, defaultHyDERetries)
162+
questions, err := runHyDEWithRetry(ctx, p.LLM, req, defaultHyDERetries, p.LLMCallTimeout)
162163
if err != nil {
163164
return nil, err
164165
}
@@ -176,7 +177,12 @@ const defaultHyDERetries = 2
176177
// parse failure returns an error so the caller can log it; transport
177178
// errors propagate. ErrNotImplemented (stub LLM) degrades to "no
178179
// questions" so test paths keep working.
179-
func runHyDEWithRetry(ctx context.Context, client llmgate.Client, baseReq llmgate.Request, maxRetries int) ([]string, error) {
180+
//
181+
// timeout bounds each individual Complete call. A per-call timeout (or
182+
// context cancellation) is terminal — it short-circuits the retry loop
183+
// rather than re-issuing a call that just hung. A non-positive timeout
184+
// disables the per-call bound.
185+
func runHyDEWithRetry(ctx context.Context, client llmgate.Client, baseReq llmgate.Request, maxRetries int, timeout time.Duration) ([]string, error) {
180186
if maxRetries < 0 {
181187
maxRetries = 0
182188
}
@@ -193,11 +199,12 @@ func runHyDEWithRetry(ctx context.Context, client llmgate.Client, baseReq llmgat
193199
}
194200
req.Messages = msgs
195201
}
196-
resp, err := client.Complete(ctx, req)
202+
resp, err := completeWithTimeout(ctx, client, req, timeout)
197203
if err != nil {
198204
// Stub clients return ErrNotImplemented — treat as "no
199205
// questions" so the pipeline proceeds without LLM access
200-
// in test setups.
206+
// in test setups. Per-call timeouts and other transport
207+
// errors propagate (the caller skips this section).
201208
if errors.Is(err, llmgate.ErrNotImplemented) {
202209
return nil, nil
203210
}

pkg/ingest/hyde_test.go

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,7 @@ func TestRunHyDEWithRetryHappy(t *testing.T) {
9191
m := &llmgate.Mock{Reply: `{"questions":["Q1","Q2","Q3","Q4","Q5"]}`}
9292
got, err := runHyDEWithRetry(context.Background(), m, llmgate.Request{
9393
Messages: []llmgate.Message{{Role: llmgate.RoleUser, Content: "go"}},
94-
}, 2)
94+
}, 2, 0)
9595
if err != nil {
9696
t.Fatalf("happy path: %v", err)
9797
}
@@ -118,7 +118,7 @@ func TestRunHyDEWithRetryRetriesOnNonJSON(t *testing.T) {
118118
}
119119
got, err := runHyDEWithRetry(context.Background(), m, llmgate.Request{
120120
Messages: []llmgate.Message{{Role: llmgate.RoleUser, Content: "go"}},
121-
}, 2)
121+
}, 2, 0)
122122
if err != nil {
123123
t.Fatalf("should recover on 3rd attempt: %v", err)
124124
}
@@ -134,7 +134,7 @@ func TestRunHyDEWithRetryFinalParseFailReturnsError(t *testing.T) {
134134
m := &llmgate.Mock{Reply: "no JSON anywhere here, just prose."}
135135
_, err := runHyDEWithRetry(context.Background(), m, llmgate.Request{
136136
Messages: []llmgate.Message{{Role: llmgate.RoleUser, Content: "go"}},
137-
}, 2)
137+
}, 2, 0)
138138
if err == nil {
139139
t.Error("want final-parse error after all retries fail")
140140
}
@@ -181,7 +181,7 @@ func TestHyDEGracefulOnNonJSON(t *testing.T) {
181181

182182
_, err := runHyDEWithRetry(context.Background(), m, llmgate.Request{
183183
Messages: []llmgate.Message{{Role: llmgate.RoleUser, Content: "u"}},
184-
}, 2)
184+
}, 2, 0)
185185
if err == nil {
186186
t.Fatal("want graceful error after 3 failed attempts")
187187
}

pkg/ingest/ingest.go

Lines changed: 66 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -157,12 +157,37 @@ type Pipeline struct {
157157
// for a table of contents. Default: 20.
158158
TOCCheckPages int
159159

160+
// LLMCallTimeout bounds each INDIVIDUAL LLM call the pipeline issues
161+
// (one section's summary, one leaf's HyDE questions, one TOC
162+
// detect/extract/verify turn). It is the safety valve against a
163+
// provider call that hangs with neither a response nor an error:
164+
// without it, that call's bounded-concurrency errgroup blocks on
165+
// Wait() forever and the document never leaves `summarizing` (observed
166+
// stuck for 13+ hours).
167+
//
168+
// A call that exceeds this deadline is handled exactly like any other
169+
// per-section failure: the surrounding stage logs it and skips the
170+
// section, leaving its existing/empty summary. One hung call can no
171+
// longer freeze the whole document.
172+
//
173+
// Zero (the Go zero value, used by Pipeline literals in tests) means
174+
// "no per-call timeout" so existing test paths that don't set it keep
175+
// their unbounded behaviour. NewPipeline defaults it to 90s.
176+
LLMCallTimeout time.Duration
177+
160178
// globalLLMSem is the lazily-initialized shared semaphore enforcing
161179
// GlobalLLMConcurrency. nil means "no global cap" — callers fall back
162180
// to per-stage limits only.
163181
globalLLMSem chan struct{}
164182
}
165183

184+
// defaultLLMCallTimeout is the per-call deadline NewPipeline applies when
185+
// LLMCallTimeout is left unset. 90s is generous for a single summarize /
186+
// HyDE / TOC turn even on a slow reasoning model, while still being short
187+
// enough that a hung call is reaped in seconds-to-low-minutes rather than
188+
// blocking the document forever.
189+
const defaultLLMCallTimeout = 90 * time.Second
190+
166191
// NewPipeline returns a Pipeline with sensible defaults filled in.
167192
func NewPipeline(p Pipeline) *Pipeline {
168193
if p.SummaryMaxChars == 0 {
@@ -210,12 +235,45 @@ func NewPipeline(p Pipeline) *Pipeline {
210235
if p.GlobalLLMConcurrency > 0 {
211236
p.globalLLMSem = make(chan struct{}, p.GlobalLLMConcurrency)
212237
}
238+
// A per-call timeout is the difference between "one bad section" and
239+
// "the whole document wedged for hours", so NewPipeline always fills
240+
// one in. Pipeline literals (test paths) that leave it zero keep the
241+
// historical unbounded behaviour.
242+
if p.LLMCallTimeout <= 0 {
243+
p.LLMCallTimeout = defaultLLMCallTimeout
244+
}
213245
if p.Logger == nil {
214246
p.Logger = slog.Default()
215247
}
216248
return &p
217249
}
218250

251+
// completeWithTimeout issues a single LLM call bounded by timeout. A
252+
// non-positive timeout disables the bound (calls ctx directly), preserving
253+
// the legacy behaviour for Pipeline literals that never set one.
254+
//
255+
// The returned error on deadline expiry is context.DeadlineExceeded
256+
// wrapped by the client — callers in the ingest stages already treat any
257+
// Complete error as a non-fatal per-section skip, so a timeout slots into
258+
// the existing degrade-and-continue path with no special-casing.
259+
func completeWithTimeout(ctx context.Context, client llmgate.Client, req llmgate.Request, timeout time.Duration) (*llmgate.Response, error) {
260+
if timeout <= 0 {
261+
return client.Complete(ctx, req)
262+
}
263+
callCtx, cancel := context.WithTimeout(ctx, timeout)
264+
defer cancel()
265+
return client.Complete(callCtx, req)
266+
}
267+
268+
// isTimeout reports whether err is a context deadline/cancellation — the
269+
// signature of a per-call LLM timeout. The ingest retry loops use it to
270+
// stop retrying immediately on a timeout: re-issuing a call that just hung
271+
// would only multiply the wall-time cost (N retries × the timeout) without
272+
// changing the outcome, so a timeout is terminal, not retryable.
273+
func isTimeout(err error) bool {
274+
return errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled)
275+
}
276+
219277
// acquireGlobalLLM blocks until a global-LLM-concurrency slot is free,
220278
// or returns false if ctx is canceled first. Returns a release func the
221279
// caller must invoke (typically deferred). Safe to call when the global
@@ -331,10 +389,11 @@ func (p *Pipeline) runTOCBuilder(ctx context.Context, docID tree.DocumentID, par
331389
model = p.SummaryModel
332390
}
333391
builder := &TOCBuilder{
334-
LLM: p.LLM,
335-
Model: model,
336-
Concurrency: p.TOCConcurrency,
337-
TOCCheckPages: p.TOCCheckPages,
392+
LLM: p.LLM,
393+
Model: model,
394+
Concurrency: p.TOCConcurrency,
395+
TOCCheckPages: p.TOCCheckPages,
396+
LLMCallTimeout: p.LLMCallTimeout,
338397
}
339398
nodes, usage, err := builder.Build(ctx, pages)
340399
if err != nil {
@@ -717,7 +776,7 @@ func (p *Pipeline) summaryFor(ctx context.Context, s db.Section, childLines []st
717776
// request. Kept for the SummaryAxesEnabled=false opt-out branch and
718777
// for unit tests that build a Pipeline literal without the new flag.
719778
func (p *Pipeline) legacyOneLineSummary(ctx context.Context, s db.Section, body, profile string) (string, error) {
720-
resp, err := p.LLM.Complete(ctx, llmgate.Request{
779+
resp, err := completeWithTimeout(ctx, p.LLM, llmgate.Request{
721780
Model: p.SummaryModel,
722781
Temperature: 0.0,
723782
MaxTokens: 260,
@@ -727,7 +786,7 @@ func (p *Pipeline) legacyOneLineSummary(ctx context.Context, s db.Section, body,
727786
"Section titled %q.\n\n%s\n\nReturn a single sentence (≤ 60 words) that names this section's concrete topics, entities, identifiers, and key items so a retrieval engine can match it to user questions.",
728787
cleanForLLM(s.Title), body)},
729788
},
730-
})
789+
}, p.LLMCallTimeout)
731790
if err != nil {
732791
// Stub LLMs return ErrNotImplemented. Degrade gracefully: use a
733792
// truncated excerpt as the "summary" so downstream retrieval
@@ -770,7 +829,7 @@ func (p *Pipeline) structuredSummaryFor(ctx context.Context, s db.Section, body,
770829
JSONSchema: []byte(summaryAxesJSONSchema),
771830
}
772831

773-
axes, rawText, err := runSummaryAxesWithRetry(ctx, p.LLM, req, defaultSummaryAxesRetries)
832+
axes, rawText, err := runSummaryAxesWithRetry(ctx, p.LLM, req, defaultSummaryAxesRetries, p.LLMCallTimeout)
774833
if err != nil {
775834
// Transport / ErrNotImplemented / unrecoverable: fall back to a
776835
// text excerpt as OneLine with empty axes. Never fail ingest.

pkg/ingest/summary_axes.go

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import (
66
"errors"
77
"fmt"
88
"strings"
9+
"time"
910

1011
"github.com/hallelx2/llmgate"
1112

@@ -82,7 +83,13 @@ const defaultSummaryAxesRetries = 2
8283
//
8384
// ErrNotImplemented (stub LLM) is folded into the error return so
8485
// callers can degrade to the text fallback path.
85-
func runSummaryAxesWithRetry(ctx context.Context, client llmgate.Client, baseReq llmgate.Request, maxRetries int) (*tree.SummaryAxes, string, error) {
86+
//
87+
// timeout bounds each individual Complete call. A per-call timeout (or
88+
// context cancellation) is terminal — it short-circuits the retry loop
89+
// and returns the error immediately, because re-issuing a call that just
90+
// hung would only multiply the wall-time cost. A non-positive timeout
91+
// disables the per-call bound.
92+
func runSummaryAxesWithRetry(ctx context.Context, client llmgate.Client, baseReq llmgate.Request, maxRetries int, timeout time.Duration) (*tree.SummaryAxes, string, error) {
8693
if maxRetries < 0 {
8794
maxRetries = 0
8895
}
@@ -99,10 +106,11 @@ func runSummaryAxesWithRetry(ctx context.Context, client llmgate.Client, baseReq
99106
}
100107
req.Messages = msgs
101108
}
102-
resp, err := client.Complete(ctx, req)
109+
resp, err := completeWithTimeout(ctx, client, req, timeout)
103110
if err != nil {
104111
// ErrNotImplemented bubbles up so the caller can use a
105-
// purely text-based fallback. Transport errors do the same.
112+
// purely text-based fallback. Transport errors and per-call
113+
// timeouts do the same (a timeout is terminal — no retry).
106114
return nil, lastRaw, err
107115
}
108116
lastRaw = resp.Content

0 commit comments

Comments
 (0)