{"record":{"id":"864225fcf93a40c9","repo":"apache/beam","slug":"failed-to-send-chunks-for-v-close-error-v","errorCode":null,"errorMessage":"failed to send chunks for %v; close error: %v","messagePattern":"failed to send chunks for (.+?); close error: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/artifact/stage.go","lineNumber":163,"sourceCode":"\n\tstream, err := client.PutArtifact(ctx)\n\tif err != nil {\n\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","sourceCodeStart":145,"sourceCodeEnd":181,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/artifact/stage.go#L145-L181","documentation":"Stage() in the Beam Go artifact package uploads a local file to the LegacyArtifactStagingService over a gRPC client stream. After the metadata header is sent, stageChunks streams 1MB data chunks; if any chunk Send fails (or reads fail), Stage tries stream.CloseAndRecv() to drain the stream and wraps the original error with the close error appended. This message means the artifact upload aborted mid-stream and reports both why the send failed and whether closing the stream also failed.","triggerScenarios":"stream.Send(chunk) inside stageChunks returns a non-EOF error (server cancelled the PutArtifact RPC, connection dropped, deadline exceeded, or the local file read failed via the r.Read error path only if not EOF). The CloseAndRecv errClose is interpolated but is secondary.","commonSituations":"Runner/artifact staging service went away mid-upload (job cancelled or crashed); network interruption or gRPC context deadline exceeded while transferring a large artifact; server rejects the session token and cancels the stream after the header; transient flakiness that MultiStage retries up to 3 times before giving up with 'failed to stage %v in 3 attempts'.","solutions":["Inspect the wrapped root cause (err) first — it names the gRPC status (Unavailable, Canceled, DeadlineExceeded) that actually killed the upload; fix that underlying condition.","Verify the artifact staging service endpoint is reachable and the staging session token is valid; an invalid token causes the server to cancel the stream after the header is sent.","Increase the gRPC context timeout / check network stability for large files; MultiStage already retries 3 times with jittered backoff, so persistent failures indicate an environment problem, not transient load.","If errClose is non-nil too, check server logs: it means the server also failed during close, typically because the stream was already terminated server-side."],"exampleFix":"// before\nstagedHash, err := stageChunks(stream, fd)\nif err != nil {\n\t_, errClose := stream.CloseAndRecv()\n\treturn nil, errors.Wrapf(err, \"failed to send chunks for %v; close error: %v\", filename, errClose)\n}\n// after (caller-side: surface and classify the root cause before retrying)\nstagedHash, err := stageChunks(stream, fd)\nif err != nil {\n\t_, errClose := stream.CloseAndRecv()\n\tif status.Code(errors.Unwrap(err)) == codes.Unavailable || status.Code(errors.Unwrap(err)) == codes.DeadlineExceeded {\n\t\treturn nil, retryable.New(errors.Wrapf(err, \"failed to send chunks for %v; close error: %v\", filename, errClose))\n\t}\n\treturn nil, errors.Wrapf(err, \"failed to send chunks for %v; close error: %v\", filename, errClose)\n}","handlingStrategy":"retry","validationCode":"conn, err := grpc.Dial(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))\nif err != nil { return err }\nclient := jobpb.NewLegacyArtifactStagingServiceClient(conn)\n// pre-check service reachability before staging\nc, cancel := context.WithTimeout(ctx, 5*time.Second)\ndefer cancel()\nif _, err := client.PutArtifact(c); err != nil {\n\treturn fmt.Errorf(\"artifact staging service unreachable: %w\", err)\n}","typeGuard":null,"tryCatchPattern":"if _, err := artifact.MultiStage(ctx, client, 10, files, token); err != nil {\n\tvar gerr errorx.GuardedError // persistent failure after 3 attempts\n\tlog.Printf(\"staging failed permanently: %v\", err)\n\t// inspect wrapped grpc status before deciding to retry at a higher level\n}","preventionTips":["Rely on MultiStage's built-in 3-attempt retry with backoff rather than calling Stage directly for flaky networks.","Set a generous context deadline proportional to artifact size.","Verify the staging session token is valid before starting a large multi-file staging run.","Monitor artifact service health/stability during pipeline submission."],"tags":["grpc","streaming","artifact-staging","network","go","beam"],"backgroundTag":"broken-pipe","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}