{"record":{"id":"354b39cc0c542758","repo":"dgraph-io/dgraph","slug":"recv-upstream-d-w","errorCode":null,"errorMessage":"recv upstream(%d): %w","messagePattern":"recv upstream\\((.+?)\\): %w","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"worker/import.go","lineNumber":407,"sourceCode":"\n\treturn pipeTwoStream(stream, alphaStream, groupId)\n}\n\nfunc pipeTwoStream(in api.Dgraph_StreamExtSnapshotServer, out pb.Worker_StreamExtSnapshotClient, groupId uint32) error {\n\tcurrentGroup := groups().Node.gid\n\tctx := in.Context()\n\n\tfor {\n\t\tif err := ctx.Err(); err != nil {\n\t\t\treturn err\n\t\t}\n\n\t\treq, err := in.Recv()\n\t\tif errors.Is(err, io.EOF) {\n\t\t\treturn nil\n\t\t}\n\t\tif err != nil {\n\t\t\treturn fmt.Errorf(\"recv upstream(%d): %w\", currentGroup, err)\n\t\t}\n\t\tif req.Pkt == nil {\n\t\t\treturn fmt.Errorf(\"unexpected empty request\")\n\t\t}\n\n\t\tif req.Pkt.Done {\n\t\t\t// Forward Done, half-close downstream send.\n\t\t\tif err := out.Send(&api.StreamExtSnapshotRequest{Pkt: req.Pkt}); err != nil && !errors.Is(err, io.EOF) {\n\t\t\t\treturn fmt.Errorf(\"send done downstream(%d): %w\", groupId, err)\n\t\t\t}\n\t\t\t_ = out.CloseSend()\n\n\t\t\t// Drain downstream and relay upstream until Finish=true.\n\t\t\tfor {\n\t\t\t\tif err := ctx.Err(); err != nil {\n\t\t\t\t\treturn err\n\t\t\t\t}\n\t\t\t\tresp, err := out.Recv()","sourceCodeStart":389,"sourceCodeEnd":425,"githubUrl":"https://github.com/dgraph-io/dgraph/blob/759e242be62c91f8d084da06ad0c8d21256d9c07/worker/import.go#L389-L425","documentation":"pipeTwoStream relays the upstream import stream (from the client) to the downstream leader. Reading from the upstream with in.Recv() failed with an error other than io.EOF, so piping aborts with this wrapped error identifying the group whose upstream read failed. It signals the import client stopped sending unexpectedly.","triggerScenarios":"in.Recv() returns a non-EOF error: the importing client (dgraph live/dgraph incremental importer) disconnected, the client context was cancelled, or a transport error occurred on the client-to-alpha stream.","commonSituations":"Client process killed or OOMed mid-snapshot; network drop between client and alpha; client hit its own timeout and cancelled the context; load balancer idle timeout closing an apparently quiet stream during slow disk flushes.","solutions":["Inspect the wrapped error: Canceled means the caller aborted, Unavailable/EOF means transport loss","Re-run the import from the client once the network is stable","Increase client-side keepalive and any intermediary idle timeouts for long streaming imports","Check client host resources (OOM killer) if the client vanished mid-stream"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// before a long import, ensure the client can hold a long-lived stream\nctx, cancel := context.WithTimeout(context.Background(), maxImportDuration)\ndefer cancel() // sized generously for the snapshot size","typeGuard":"func isUpstreamDisconnect(err error) bool {\n    return status.Code(err) == codes.Canceled || status.Code(err) == codes.Unavailable || errors.Is(err, io.EOF)\n}","tryCatchPattern":"if err != nil && strings.Contains(err.Error(), \"recv upstream\") {\n    if isUpstreamDisconnect(errors.Unwrap(err)) {\n        // client dropped; restart the import from the client\n    }\n}","preventionTips":["Run import clients with sufficient memory to avoid OOM kills","Keep the client process alive for the whole transfer","Raise intermediary idle timeouts for long streams","Add gRPC keepalive on the client"],"tags":["grpc","streaming","io","relay"],"backgroundTag":"upstream-stream-recv-failed","analyzedSha":"759e242be62c91f8d084da06ad0c8d21256d9c07","analyzedAt":"2026-09-01T14:42:12.034Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}