{"record":{"id":"5baff14dae36dade","repo":"apache/beam","slug":"failed-to-send-staging-token","errorCode":null,"errorMessage":"failed to send staging token","messagePattern":"failed to send staging token","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/universal/runnerlib/stage.go","lineNumber":86,"sourceCode":"\t\t\treturn errors.Errorf(\"failed to stage artifacts for token %v in %v attempts: %v\", st, attempts, strings.Join(failures, \";\\n\"))\n\t\t}\n\t}\n}\n\nfunc stageFiles(ctx context.Context, cc *grpc.ClientConn, binary, st string) error {\n\tclient := jobpb.NewArtifactStagingServiceClient(cc)\n\tstream, err := client.ReverseArtifactRetrievalService(ctx)\n\tif err != nil {\n\t\treturn err\n\t}\n\tdefer func() {\n\t\tif err := stream.CloseSend(); err != nil {\n\t\t\tlog.Error(ctx, \"StageViaPortableApi CloseSend error: \", err)\n\t\t}\n\t}()\n\n\tif err := stream.Send(&jobpb.ArtifactResponseWrapper{StagingToken: st}); err != nil {\n\t\treturn errors.Wrapf(err, \"failed to send staging token\")\n\t}\n\n\tfor {\n\t\tin, err := stream.Recv()\n\t\tif err == io.EOF {\n\t\t\treturn nil\n\t\t}\n\t\tif err != nil {\n\t\t\treturn err\n\t\t}\n\n\t\tswitch request := in.Request.(type) {\n\t\tcase *jobpb.ArtifactRequestWrapper_ResolveArtifact:\n\t\t\terr = stream.Send(&jobpb.ArtifactResponseWrapper{\n\t\t\t\tResponse: &jobpb.ArtifactResponseWrapper_ResolveArtifactResponse{\n\t\t\t\t\tResolveArtifactResponse: &jobpb.ResolveArtifactsResponse{\n\t\t\t\t\t\tReplacements: request.ResolveArtifact.Artifacts,\n\t\t\t\t\t},","sourceCodeStart":68,"sourceCodeEnd":104,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/universal/runnerlib/stage.go#L68-L104","documentation":"stageFiles wraps the gRPC Send error when it fails to send the initial ArtifactResponseWrapper carrying the staging token on the ReverseArtifactRetrievalService stream. It indicates the artifact staging stream could not even be established with the token.","triggerScenarios":"stream.Send(&jobpb.ArtifactResponseWrapper{StagingToken: st}) returns a gRPC error — connection dropped, server rejected the stream, or the artifact service does not implement reverse artifact retrieval.","commonSituations":"Artifact service version doesn't implement ReverseArtifactRetrievalService, TLS/plaintext mismatch on the connection, connection reset because the job server is overloaded or shut down.","solutions":["Check the wrapped gRPC error for connection status (Unavailable, Unimplemented)","If Unimplemented, upgrade the job server to a version supporting reverse artifact retrieval","Verify transport security settings match the server (TLS vs plaintext)","Confirm network connectivity to the artifact endpoint"],"exampleFix":"// before\nif err := stream.Send(&jobpb.ArtifactResponseWrapper{StagingToken: st}); err != nil {\n  return errors.Wrapf(err, \"failed to send staging token\")\n}\n// after: diagnose via the wrapped status, e.g. status.Code(err) == codes.Unimplemented -> upgrade job server","handlingStrategy":"retry","validationCode":"status := statusFromErr(err) // check codes.Unavailable/Unimplemented after failure\n// pre-check: ensure job server supports reverse artifact retrieval","typeGuard":null,"tryCatchPattern":"if err := stageFiles(ctx, cc, binary, st); err != nil {\n  if status.Code(errors.Unwrap(err)) == codes.Unimplemented {\n    // job server too old: upgrade job server\n  }\n}","preventionTips":["Use a job server version that implements ReverseArtifactRetrievalService","Match TLS/plaintext settings between client dial options and server","Monitor connection stability under load for large artifacts"],"tags":["grpc","artifact-staging","network","staging-token"],"backgroundTag":"broken-pipe","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}