{"record":{"id":"a004152c929aee6a","repo":"dgraph-io/dgraph","slug":"send-done-downstream-d-w","errorCode":null,"errorMessage":"send done downstream(%d): %w","messagePattern":"send done downstream\\((.+?)\\): %w","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"worker/import.go","lineNumber":416,"sourceCode":"\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()\n\t\t\t\tif errors.Is(err, io.EOF) {\n\t\t\t\t\treturn fmt.Errorf(\"downstream(%d) closed before Finish=true\", groupId)\n\t\t\t\t}\n\t\t\t\tif err != nil {\n\t\t\t\t\treturn fmt.Errorf(\"recv final downstream(%d): %w\", groupId, err)\n\t\t\t\t}\n\t\t\t\tif err := in.Send(resp); err != nil {\n\t\t\t\t\treturn fmt.Errorf(\"relay final upstream: %w\", err)\n\t\t\t\t}","sourceCodeStart":398,"sourceCodeEnd":434,"githubUrl":"https://github.com/dgraph-io/dgraph/blob/759e242be62c91f8d084da06ad0c8d21256d9c07/worker/import.go#L398-L434","documentation":"When the upstream sends the Done packet, pipeTwoStream forwards it downstream and half-closes its send side, then waits for the leader to finish. If that out.Send of the Done packet fails (and it is not a benign io.EOF), the relay aborts with this wrapped error naming the downstream group.","triggerScenarios":"out.Send(&api.StreamExtSnapshotRequest{Pkt: req.Pkt}) errors because the gRPC client stream to the leader was already closed/broken, the leader errored and tore the stream down, or the shared context was cancelled.","commonSituations":"Leader crashed or was restarted just as the snapshot transfer finished; long import outlived an intermediary's connection timeout; leader returned an error earlier and closed the stream, making the final Done send fail.","solutions":["Check the wrapped error and the leader's logs for the server-side failure that closed the stream","Retry the import after confirming leader health","Keep the whole import under any proxy idle/timeout limits or disable them for internal gRPC traffic","Treat an io.EOF here as benign — the code deliberately ignores it"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// precheck leader disk/memory headroom before finalizing a snapshot\ndfEnoughDisk(leaderHost, requiredBytes); memEnough(leaderHost, requiredBytes)","typeGuard":"func isBenignEOF(err error) bool { return errors.Is(err, io.EOF) } // code already treats this as benign","tryCatchPattern":"if err != nil && strings.Contains(err.Error(), \"send done downstream\") {\n    // leader stream broke during half-close; check leader logs, retry import\n}","preventionTips":["Keep the leader healthy through the entire import window","Disable idle timeouts that fire during final flushes","Retry failed imports; Done-phase failures leave state incomplete","Monitor leader restarts (OOM, upgrades) during imports"],"tags":["grpc","streaming","half-close","relay"],"backgroundTag":"downstream-stream-send-failed","analyzedSha":"759e242be62c91f8d084da06ad0c8d21256d9c07","analyzedAt":"2026-09-01T14:42:12.034Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}