{"record":{"id":"651341574258969c","repo":"vxcontrol/pentagi","slug":"flow-d-input-processing-timeout-w","errorCode":null,"errorMessage":"flow %d input processing timeout: %w","messagePattern":"flow (.+?) input processing timeout: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"backend/pkg/controller/flow.go","lineNumber":667,"sourceCode":"\tctx, span := obs.Observer.NewSpan(ctx, obs.SpanKindInternal, \"controller.flowWorker.PutInput\")\n\tdefer span.End()\n\n\tif err := fw.switchProvider(ctx, prv); err != nil {\n\t\treturn fmt.Errorf(\"failed to switch provider: %w\", err)\n\t}\n\n\tif err := fw.PutResources(ctx, resources); err != nil {\n\t\tfw.logger.WithError(err).Warn(\"failed to copy resources before user input\")\n\t}\n\n\tflin := flowInput{input: input, done: make(chan error, 1)}\n\tselect {\n\tcase <-fw.ctx.Done():\n\t\tclose(flin.done)\n\t\treturn fmt.Errorf(\"flow %d stopped: %w\", fw.flowCtx.FlowID, fw.ctx.Err())\n\tcase <-ctx.Done():\n\t\tclose(flin.done)\n\t\treturn fmt.Errorf(\"flow %d input processing timeout: %w\", fw.flowCtx.FlowID, ctx.Err())\n\tcase fw.input <- flin:\n\t\ttimer := time.NewTimer(flowInputTimeout)\n\t\tdefer timer.Stop()\n\n\t\tselect {\n\t\tcase err := <-flin.done:\n\t\t\treturn err\n\t\tcase <-timer.C:\n\t\t\treturn nil // no early error\n\t\tcase <-fw.ctx.Done():\n\t\t\treturn fmt.Errorf(\"flow %d stopped: %w\", fw.flowCtx.FlowID, fw.ctx.Err())\n\t\tcase <-ctx.Done():\n\t\t\treturn fmt.Errorf(\"flow %d input processing timeout: %w\", fw.flowCtx.FlowID, ctx.Err())\n\t\t}\n\t}\n}\n\nfunc (fw *flowWorker) PutResources(ctx context.Context, dbResources []database.UserResource) error {","sourceCodeStart":649,"sourceCodeEnd":685,"githubUrl":"https://github.com/vxcontrol/pentagi/blob/ea665308baaff015b226f308438a68d929d0f29b/backend/pkg/controller/flow.go#L649-L685","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Pass a context with a timeout comfortably longer than the flow's expected step duration.","Retry the PutInput when the worker is less busy; the input was never delivered so it is safe to resend.","Consider asynchronous input submission (queue + subscription) instead of blocking on a request-scoped ctx.","Increase flow worker throughput or drain fw.input faster if timeouts recur."],"exampleFix":"// before\nctx, cancel := context.WithTimeout(ctx, 2*time.Second)\nerr := fw.PutInput(ctx, prv, input, res) // worker busy -> timeout\n// after\nctx, cancel := context.WithTimeout(ctx, 60*time.Second)\nerr := fw.PutInput(ctx, prv, input, res)","handlingStrategy":"retry","validationCode":"select {\ncase <-ctx.Done():\n    return fmt.Errorf(\"caller ctx done before submit: %w\", ctx.Err())\ndefault:\n}\n// ensure remaining ctx budget exceeds expected enqueue wait","typeGuard":"func isCallerTimeoutErr(err error) bool {\n    return err != nil && strings.Contains(err.Error(), \"input processing timeout\") &&\n        errors.Is(err, context.DeadlineExceeded)\n}","tryCatchPattern":"err := worker.PutInput(ctx, prv, input, res)\nif isCallerTimeoutErr(err) {\n    // input never delivered; safe to retry with a longer deadline\n    ctx2, cancel := context.WithTimeout(context.Background(), 60*time.Second)\n    defer cancel()\n    return worker.PutInput(ctx2, prv, input, res)\n}","preventionTips":["Size request timeouts larger than expected worker busy periods.","Submit input asynchronously (queue + subscription) instead of blocking a request ctx.","Retry PutInput on caller-timeout since the input was not enqueued.","Monitor fw.input channel depth to detect chronic backpressure."],"tags":["go","timeout","context","backpressure"],"backgroundTag":"context-deadline-exceeded","analyzedSha":"ea665308baaff015b226f308438a68d929d0f29b","analyzedAt":"2026-09-01T14:16:31.421Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}