vxcontrol/pentagi · warning

flow %d input processing timeout: %w

Error message

flow %d input processing timeout: %w

What it means

PutInput returns this when the caller's ctx is cancelled or times out while the code waits to enqueue the input onto fw.input — the flow worker's input channel is full (worker busy) and the caller gave up before the value was delivered. Distinct from error 93: here it is the CALLER's context that expired, not the worker's.

Source

Thrown at backend/pkg/controller/flow.go:667

	ctx, span := obs.Observer.NewSpan(ctx, obs.SpanKindInternal, "controller.flowWorker.PutInput")
	defer span.End()

	if err := fw.switchProvider(ctx, prv); err != nil {
		return fmt.Errorf("failed to switch provider: %w", err)
	}

	if err := fw.PutResources(ctx, resources); err != nil {
		fw.logger.WithError(err).Warn("failed to copy resources before user input")
	}

	flin := flowInput{input: input, done: make(chan error, 1)}
	select {
	case <-fw.ctx.Done():
		close(flin.done)
		return fmt.Errorf("flow %d stopped: %w", fw.flowCtx.FlowID, fw.ctx.Err())
	case <-ctx.Done():
		close(flin.done)
		return fmt.Errorf("flow %d input processing timeout: %w", fw.flowCtx.FlowID, ctx.Err())
	case fw.input <- flin:
		timer := time.NewTimer(flowInputTimeout)
		defer timer.Stop()

		select {
		case err := <-flin.done:
			return err
		case <-timer.C:
			return nil // no early error
		case <-fw.ctx.Done():
			return fmt.Errorf("flow %d stopped: %w", fw.flowCtx.FlowID, fw.ctx.Err())
		case <-ctx.Done():
			return fmt.Errorf("flow %d input processing timeout: %w", fw.flowCtx.FlowID, ctx.Err())
		}
	}
}

func (fw *flowWorker) PutResources(ctx context.Context, dbResources []database.UserResource) error {

View on GitHub (pinned to ea665308ba)

Solutions

1. Pass a context with a timeout comfortably longer than the flow's expected step duration.
2. Retry the PutInput when the worker is less busy; the input was never delivered so it is safe to resend.
3. Consider asynchronous input submission (queue + subscription) instead of blocking on a request-scoped ctx.
4. Increase flow worker throughput or drain fw.input faster if timeouts recur.

Example fix

// before
ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
err := fw.PutInput(ctx, prv, input, res) // worker busy -> timeout
// after
ctx, cancel := context.WithTimeout(ctx, 60*time.Second)
err := fw.PutInput(ctx, prv, input, res)

select {
case <-ctx.Done():
    return fmt.Errorf("caller ctx done before submit: %w", ctx.Err())
default:
}
// ensure remaining ctx budget exceeds expected enqueue wait

func isCallerTimeoutErr(err error) bool {
    return err != nil && strings.Contains(err.Error(), "input processing timeout") &&
        errors.Is(err, context.DeadlineExceeded)
}

err := worker.PutInput(ctx, prv, input, res)
if isCallerTimeoutErr(err) {
    // input never delivered; safe to retry with a longer deadline
    ctx2, cancel := context.WithTimeout(context.Background(), 60*time.Second)
    defer cancel()
    return worker.PutInput(ctx2, prv, input, res)
}

When it happens

Trigger: Calling PutInput with a short-lived request context while the flow worker is busy processing a previous step; the select takes ctx.Done() before fw.input accepts the flowInput.

Common situations: HTTP request timeout (e.g. Gin write timeout) shorter than the worker's current task duration; a client disconnect cancels the request ctx; nested context.WithTimeout too aggressive for long-running flows.

Understand the failure class

• Timeouts: ETIMEDOUT, deadlines, and hung requests — what actually expires when a request times out.

Related errors

• task stop timeout
• listing container directory '%s': %w
• wait cancelled — the assistant session was interrupted while
• bearer scheme must be used
• token can't be empty

AI-assisted analysis of vxcontrol/pentagi@ea665308ba (2026-09-01). Data as JSON: /api/errors/651341574258969c. Report an issue: GitHub.