{"record":{"id":"f237e47a9479a791","repo":"apache/beam","slug":"failed-to-close-stream-for-v-response-v","errorCode":null,"errorMessage":"failed to close stream for %v; response: %v","messagePattern":"failed to close stream for (.+?); response: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/artifact/stage.go","lineNumber":166,"sourceCode":"\t\treturn nil, err\n\t}\n\n\theader := &jobpb.PutArtifactRequest{\n\t\tContent: &jobpb.PutArtifactRequest_Metadata{\n\t\t\tMetadata: pmd,\n\t\t},\n\t}\n\tif err := stream.Send(header); err != nil {\n\t\tstream.CloseAndRecv() // ignore error\n\t\treturn nil, errors.Wrapf(err, \"failed to send header for %v\", filename)\n\t}\n\tstagedHash, err := stageChunks(stream, fd)\n\tif err != nil {\n\t\t_, errClose := stream.CloseAndRecv()\n\t\treturn nil, errors.Wrapf(err, \"failed to send chunks for %v; close error: %v\", filename, errClose)\n\t}\n\tif resp, err := stream.CloseAndRecv(); err != nil && err != io.EOF {\n\t\treturn nil, errors.Wrapf(err, \"failed to close stream for %v; response: %v\", filename, resp)\n\t}\n\tif hash != stagedHash {\n\t\treturn nil, errors.Errorf(\"unexpected SHA256 for sent chunks for %v: %v, want %v\", filename, stagedHash, hash)\n\t}\n\treturn md, nil\n}\n\nfunc stageChunks(stream jobpb.LegacyArtifactStagingService_PutArtifactClient, r io.Reader) (string, error) {\n\tsha256W := sha256.New()\n\tdata := make([]byte, 1<<20)\n\tfor {\n\t\tn, err := r.Read(data)\n\t\tif n > 0 {\n\t\t\tif _, err := sha256W.Write(data[:n]); err != nil {\n\t\t\t\tpanic(err) // cannot fail\n\t\t\t}\n\n\t\t\tchunk := &jobpb.PutArtifactRequest{","sourceCodeStart":148,"sourceCodeEnd":184,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/artifact/stage.go#L148-L184","documentation":"Stage() closes the PutArtifact gRPC stream with CloseAndRecv() to get the server's final response. If the close/receive returns an error other than io.EOF, Stage wraps it as 'failed to close stream for %v; response: %v'. This means the server did not cleanly finish the artifact upload — the final RPC handshake failed even though all chunks may have been sent.","triggerScenarios":"stream.CloseAndRecv() after successful chunk sends returns a non-EOF error: server returned a gRPC status error (e.g. InvalidArgument, permission/token check failure at commit time, Unavailable), the context was cancelled, or the connection was reset before the server's response arrived. resp (usually nil on error) is interpolated into the message.","commonSituations":"Staging session token rejected when the server finalizes the artifact; artifact service crashed or was redeployed between chunk upload and close; per-request context deadline exceeded for very large artifacts; job pipeline cancelled while staging was in flight.","solutions":["Check the wrapped gRPC status code: io.EOF is treated as success here, any other status is a real server-side rejection or transport failure.","Verify the staging session token is valid and not expired — servers commonly reject at CloseAndRecv time when finalizing.","Increase the context deadline passed to Stage/MultiStage for large artifacts, or reduce artifact size.","Confirm the artifact staging service is healthy and stable for the whole upload duration; retry via MultiStage (built-in 3 attempts) only for transient statuses like Unavailable."],"exampleFix":"// before\nif resp, err := stream.CloseAndRecv(); err != nil && err != io.EOF {\n\treturn nil, errors.Wrapf(err, \"failed to close stream for %v; response: %v\", filename, resp)\n}\n// after (caller-side: check gRPC status to distinguish transient vs fatal)\nif resp, err := stream.CloseAndRecv(); err != nil && err != io.EOF {\n\tswitch status.Code(errors.Unwrap(err)) {\n\tcase codes.Unavailable, codes.DeadlineExceeded:\n\t\treturn nil, retryable.New(errors.Wrapf(err, \"failed to close stream for %v; response: %v\", filename, resp))\n\tdefault:\n\t\treturn nil, errors.Wrapf(err, \"failed to close stream for %v; response: %v\", filename, resp)\n\t}\n}","handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"if _, err := artifact.Stage(ctx, client, key, filename, token); err != nil {\n\tif code := status.Code(errors.Unwrap(err)); code == codes.Unavailable || code == codes.DeadlineExceeded {\n\t\t// transient: retry with backoff\n\t} else {\n\t\treturn fmt.Errorf(\"staging close rejected (code=%v): %w\", code, err)\n\t}\n}","preventionTips":["Pass a context with sufficient deadline for the full upload plus server-side finalization.","Ensure the staging session token stays valid until all artifacts are closed.","Avoid cancelling the pipeline context while staging is in flight.","Check artifact service logs if CloseAndRecv consistently fails for specific artifact types."],"tags":["grpc","streaming","artifact-staging","rpc-close","go","beam"],"backgroundTag":"api-error-response","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"}