{"record":{"id":"2b994d7182c17a4a","repo":"apache/beam","slug":"control-recv-failed","errorCode":null,"errorMessage":"control.Recv failed","messagePattern":"control\\.Recv failed","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/harness/harness.go","lineNumber":200,"sourceCode":"\t\t}\n\t}\n\n\t// gRPC requires all readers of a stream be the same goroutine, so this goroutine\n\t// is responsible for managing the network data. All it does is pull data from\n\t// the stream, and hand off the message to a goroutine to actually be handled,\n\t// so as to avoid blocking the underlying network channel.\n\tvar shutdown int32\n\tfor {\n\t\treq, err := stub.Recv()\n\t\tif err != nil {\n\t\t\t// An error means we can't send or receive anymore. Shut down.\n\t\t\tatomic.AddInt32(&shutdown, 1)\n\t\t\tclose(respc)\n\t\t\twg.Wait()\n\t\t\tif err == io.EOF {\n\t\t\t\treturn nil\n\t\t\t}\n\t\t\treturn errors.Wrapf(err, \"control.Recv failed\")\n\t\t}\n\n\t\t// Launch a goroutine to handle the control message.\n\t\tfn := func(ctx context.Context, req *fnpb.InstructionRequest) {\n\t\t\t// TODO(lostluck): 2023/03/29 fix debug level logging to be flagged.\n\t\t\t// log.Debugf(ctx, \"RECV: %v\", proto.MarshalTextString(req))\n\t\t\tctx = hooks.RunRequestHooks(ctx, req)\n\t\t\tresp := ctrl.handleInstruction(ctx, req)\n\n\t\t\thooks.RunResponseHooks(ctx, req, resp)\n\n\t\t\tif resp != nil && atomic.LoadInt32(&shutdown) == 0 {\n\t\t\t\trespc <- resp\n\t\t\t}\n\t\t}\n\n\t\tif req.GetProcessBundle() != nil {\n\t\t\t// Add this to the inactive queue before allowing other requests","sourceCodeStart":182,"sourceCodeEnd":218,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/harness/harness.go#L182-L218","documentation":"The harness receives instruction requests from the control stream in a loop via stub.Recv(). On any error other than a clean io.EOF (which means the server ended the stream normally, e.g. graceful shutdown), the loop exits and wraps the error as \"control.Recv failed\". It indicates the control stream broke mid-run, typically due to a network interruption or a server-side failure while executing bundles.","triggerScenarios":"stub.Recv() returns a non-EOF error during the receive loop: stream reset, connection dropped, server crashed, context cancelled abnormally, or a gRPC deadline exceeded while waiting for the next instruction.","commonSituations":"Long-running pipelines where the control connection is dropped by load-balancer idle timeouts; runner master restarts mid-job; network partition between worker and runner; OOM-killed control server during heavy bundles.","solutions":["Inspect the wrapped gRPC status to distinguish Canceled (normal shutdown), Unavailable/DeadlineExceeded (network), or Internal (server crash).","Enable gRPC keepalives and raise LB idle timeouts so long-quiet streams are not dropped.","Ensure the runner/control service has enough memory; check for OOM kills around the failure time.","Re-run the pipeline; if it fails at the same bundle, investigate that bundle/transform for server-side crashes.","Upgrade runner+SDK to versions with resilient reconnect handling for the control stream."],"exampleFix":"// server-side: keep long-lived streams alive\nsrv := grpc.NewServer(\n    grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{\n        MinTime: 10 * time.Second, PermitWithoutStream: true,\n    }),\n)\n// harness side: ensure ctx is not cancelled prematurely and idle timeouts exceed max bundle duration","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"if err := runHarness(ctx, opts); err != nil {\n    if strings.Contains(err.Error(), \"control.Recv failed\") {\n        code := status.Code(errors.Unwrap(err))\n        if code == codes.Canceled {\n            return nil // graceful shutdown path\n        }\n        // otherwise: connection dropped mid-run — alert/inspect network and server health\n    }\n    return err\n}","preventionTips":["Configure gRPC keepalives and ensure LB idle timeouts exceed the longest quiet period between instructions.","Monitor control-server memory to prevent OOM kills mid-stream.","Pin matching runner/SDK versions; newer releases include stream resilience fixes.","Treat Canceled during shutdown as normal; alert on Unavailable/Internal."],"tags":["network","grpc","beam-go","stream","runtime"],"backgroundTag":"grpc-stream-broken","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}