{"record":{"id":"208c188b536e8b1c","repo":"dgraph-io/dgraph","slug":"send-data-downstream-d-w","errorCode":null,"errorMessage":"send data downstream(%d): %w","messagePattern":"send data downstream\\((.+?)\\): %w","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"worker/import.go","lineNumber":444,"sourceCode":"\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}\n\t\t\t\tif resp.Finish {\n\t\t\t\t\tglog.Infof(\"[import] [forward %d -> %d] finish\", currentGroup, groupId)\n\t\t\t\t\treturn nil\n\t\t\t\t}\n\t\t\t}\n\t\t}\n\n\t\t// Normal data chunk: send -> wait ack -> send upstream ack.\n\t\tif err := out.Send(&api.StreamExtSnapshotRequest{Pkt: req.Pkt}); err != nil {\n\t\t\treturn fmt.Errorf(\"send data downstream(%d): %w\", groupId, err)\n\t\t}\n\t\tif _, err := out.Recv(); err != nil {\n\t\t\treturn fmt.Errorf(\"ack data downstream(%d): %w\", groupId, err)\n\t\t}\n\t\tif err := in.Send(&api.StreamExtSnapshotResponse{}); err != nil {\n\t\t\treturn fmt.Errorf(\"send ack upstream: %w\", err)\n\t\t}\n\n\t}\n}\n\nfunc (w *grpcWorker) UpdateExtSnapshotStreamingState(ctx context.Context,\n\treq *api.UpdateExtSnapshotStreamingStateRequest) (*pb.Status, error) {\n\tif req == nil {\n\t\treturn nil, errors.New(\"UpdateExtSnapshotStreamingStateRequest must not be nil\")\n\t}\n\n\tif req.Start && req.Finish {","sourceCodeStart":426,"sourceCodeEnd":462,"githubUrl":"https://github.com/dgraph-io/dgraph/blob/759e242be62c91f8d084da06ad0c8d21256d9c07/worker/import.go#L426-L462","documentation":"For each normal data chunk, pipeTwoStream forwards the packet to the leader, waits for the leader's ack via out.Recv(), and only then acks upstream. If the downstream send of a data packet fails, the ack chain is broken and the relay aborts with this wrapped error naming the group.","triggerScenarios":"out.Send(&api.StreamExtSnapshotRequest{Pkt: req.Pkt}) errors mid-transfer: connection to the leader dropped, the leader cancelled or errored its stream, or gRPC flow control stalled until the context deadline hit.","commonSituations":"Leader crash mid-transfer; network instability between alphas during a large snapshot; connection-killing middleboxes during long quiet ack periods; leader-side backpressure exceeding client keepalive timeouts.","solutions":["Check the leader's logs for the failure that broke the stream, then retry the import","Verify direct, stable connectivity (and keepalives) between the two alphas","Ensure the leader has disk/memory headroom to keep acking under load","Exclude internal gRPC ports from proxy idle timeouts"],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// stable path + leader health check before starting bulk transfer\npingLeader(leaderAddr); checkKeepaliveConfig(grpcDialOpts)","typeGuard":"func isDownstreamBreak(err error) bool {\n    c := status.Code(err)\n    return errors.Is(err, io.EOF) || c == codes.Unavailable || c == codes.Canceled\n}","tryCatchPattern":"if err != nil && strings.Contains(err.Error(), \"send data downstream\") {\n    // mid-transfer break; check leader logs/network, then re-run import\n}","preventionTips":["Exclude internal gRPC ports from proxy idle/timeout handling","Configure keepalive on both alphas for long transfers","Monitor leader disk/memory so acks keep flowing under load","Schedule imports away from leader restarts/upgrades"],"tags":["grpc","streaming","data-transfer","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"}