{"record":{"id":"47dc0ef9d875e0e3","repo":"apache/beam","slug":"chunk-send-failed-stage","errorCode":null,"errorMessage":"chunk send failed","messagePattern":"chunk send failed","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/artifact/stage.go","lineNumber":196,"sourceCode":"\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{\n\t\t\t\tContent: &jobpb.PutArtifactRequest_Data{\n\t\t\t\t\tData: &jobpb.ArtifactChunk{\n\t\t\t\t\t\tData: data[:n],\n\t\t\t\t\t},\n\t\t\t\t},\n\t\t\t}\n\t\t\terr := stream.Send(chunk)\n\t\t\tif err == io.EOF {\n\t\t\t\treturn \"\", err\n\t\t\t}\n\t\t\tif err != nil {\n\t\t\t\treturn \"\", errors.Wrap(err, \"chunk send failed\")\n\t\t\t}\n\t\t}\n\t\tif err == io.EOF {\n\t\t\tbreak\n\t\t}\n\t\tif err != nil {\n\t\t\treturn \"\", err\n\t\t}\n\t}\n\treturn hex.EncodeToString(sha256W.Sum(nil)), nil\n}\n\n// KeyedFile is a key and filename pair.\ntype KeyedFile struct {\n\tKey, Filename string\n}\n\nfunc scan(dir string) ([]KeyedFile, error) {","sourceCodeStart":178,"sourceCodeEnd":214,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/artifact/stage.go#L178-L214","documentation":"stageChunks reads the local file in 1MB chunks and sends each as a PutArtifactRequest over the gRPC stream. If stream.Send returns io.EOF the stream was closed by the server (returned verbatim); any other non-nil error is wrapped as 'chunk send failed'. This indicates the gRPC client stream to the artifact staging service broke while transferring artifact data.","triggerScenarios":"stream.Send(chunk) returns a non-EOF error mid-upload: server cancelled the PutArtifact RPC (invalid staging session token, metadata rejection), transport failure/connection reset, context deadline or cancellation, or message size limits on very large files.","commonSituations":"Network partition between the runner and the artifact service; artifact service pod restart (e.g. Kubernetes) mid-upload; context timeout too short for large files; server-side cancellation because the staging session token was revoked or the manifest commit already happened.","solutions":["Read the wrapped gRPC status from the error (errors.Unwrap + status.Code) to distinguish transient (Unavailable, DeadlineExceeded) from fatal (InvalidArgument, Unauthenticated) causes.","Let MultiStage retry — it retries Stage 3 times with jittered backoff for transient failures; persistent failures need the root cause fixed.","Increase the context timeout for large artifacts and confirm no proxy/firewall is killing long-lived gRPC streams (e.g. idle/connection timeouts).","Verify the staging session token is valid for the duration of the upload."],"exampleFix":"// before\nerr := stream.Send(chunk)\nif err == io.EOF {\n\treturn \"\", err\n}\nif err != nil {\n\treturn \"\", errors.Wrap(err, \"chunk send failed\")\n}\n// after (caller-side: retry only transient send failures)\nvar lastErr error\nfor attempt := 0; attempt < 3; attempt++ {\n\t_, err := artifact.Stage(ctx, client, key, filename, token)\n\tif err == nil {\n\t\tbreak\n\t}\n\tif st, ok := status.FromError(errors.Unwrap(err)); ok &&\n\t\t(st.Code() == codes.Unavailable || st.Code() == codes.DeadlineExceeded) {\n\t\tlastErr = err\n\t\ttime.Sleep(time.Duration(attempt+1) * time.Second)\n\t\tcontinue\n\t}\n\treturn err\n}\n_ = lastErr","handlingStrategy":"retry","validationCode":"// pre-check connectivity and context budget\nif ctx.Err() != nil { return ctx.Err() }\nif _, err := grpc.DialContext(ctx, addr, grpc.WithBlock(), grpc.WithTransportCredentials(insecure.NewCredentials())); err != nil {\n\treturn fmt.Errorf(\"cannot reach artifact staging service: %w\", err)\n}","typeGuard":null,"tryCatchPattern":"_, err := artifact.Stage(ctx, client, key, filename, token)\nif err != nil && strings.Contains(err.Error(), \"chunk send failed\") {\n\tif st, ok := status.FromError(errors.Unwrap(err)); ok && (st.Code() == codes.Unavailable || st.Code() == codes.DeadlineExceeded) {\n\t\t// retry with exponential backoff; MultiStage does this internally\n\t}\n}","preventionTips":["Use MultiStage instead of raw Stage to get automatic 3-attempt retries.","Size context timeouts to artifact size (1MB chunks; large jars need minutes).","Check proxy/LB idle timeouts that kill long gRPC streams.","Keep the staging session token valid for the whole upload."],"tags":["grpc","streaming","send-failed","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"}