{"record":{"id":"ac37118ae7a34e09","repo":"dgraph-io/dgraph","slug":"failed-to-send-data-chunk-w","errorCode":null,"errorMessage":"failed to send data chunk: %w","messagePattern":"failed to send data chunk: %w","errorType":"console","errorClass":null,"httpStatus":null,"severity":"error","filePath":"dgraph/cmd/dgraphimport/import_client.go","lineNumber":222,"sourceCode":"\t// Configure and start the BadgerDB stream\n\tglog.Infof(\"[import] Starting BadgerDB stream for group [%v]\", groupId)\n\tif err := streamBadger(ctx, ps, out, groupId); err != nil {\n\t\treturn fmt.Errorf(\"badger streaming failed for group [%v]: %v\", groupId, err)\n\t}\n\treturn nil\n}\n\n// streamBadger runs a BadgerDB stream to send key-value pairs to the specified group.\n// It creates a new stream at the maximum sequence number and sends the data to the specified group.\n// It also sends a final 'done' signal to mark completion.\nfunc streamBadger(ctx context.Context, ps *badger.DB, out api.Dgraph_StreamExtSnapshotClient, groupId uint32) error {\n\tstream := ps.NewStreamAt(math.MaxUint64)\n\tstream.LogPrefix = \"[import] Sending external snapshot to group [\" + fmt.Sprintf(\"%d\", groupId) + \"]\"\n\tstream.KeyToList = nil\n\tstream.Send = func(buf *z.Buffer) error {\n\t\tp := &api.StreamPacket{Data: buf.Bytes()}\n\t\tif err := out.Send(&api.StreamExtSnapshotRequest{Pkt: p}); err != nil && !errors.Is(err, io.EOF) {\n\t\t\treturn fmt.Errorf(\"failed to send data chunk: %w\", err)\n\t\t}\n\t\tif _, err := out.Recv(); err != nil {\n\t\t\treturn fmt.Errorf(\"failed to receive response for group ID [%v] from the server: %w\", groupId, err)\n\t\t}\n\t\tglog.Infof(\"[import] Group [%v]: Received ACK for sending data chunk\", groupId)\n\n\t\treturn nil\n\t}\n\n\t// Execute the stream process\n\tif err := stream.Orchestrate(ctx); err != nil {\n\t\treturn fmt.Errorf(\"stream orchestration failed for group [%v]: %w, badger path: %s\", groupId, err, ps.Opts().Dir)\n\t}\n\n\t// Send the final 'done' signal to mark completion\n\tglog.Infof(\"[import] Sending completion signal for group [%d]\", groupId)\n\tdone := &api.StreamPacket{Done: true}\n","sourceCodeStart":204,"sourceCodeEnd":240,"githubUrl":"https://github.com/dgraph-io/dgraph/blob/759e242be62c91f8d084da06ad0c8d21256d9c07/dgraph/cmd/dgraphimport/import_client.go#L204-L240","documentation":"Produced inside the badger stream.Send callback when out.Send of a StreamPacket fails with a non-EOF error. Each packet must be sent and ACKed; a send failure aborts Orchestrate with this wrapped error. EOF is deliberately tolerated because the server may close its send side.","triggerScenarios":"gRPC stream broken mid-transfer (connection reset, server crash, LB timeout); context cancelled while blocking on Send; server closed the stream after an earlier error so subsequent Sends fail.","commonSituations":"Multi-GB snapshot transfer interrupted by network hiccup; k8s service/ingress killing long-lived streams; Alpha pod evicted mid-import.","solutions":["Check the wrapped gRPC status for Unavailable/Canceled to confirm a transport break.","Verify server health (pod restarts, OOM kills) and network path stability.","Raise keepalive/timeout settings on client and LB so long streams survive.","Re-run the import; streams cannot resume mid-transfer.","Reduce packet pressure only if server logs show backpressure/OOM at that time."],"exampleFix":"// before\nif err := out.Send(&api.StreamExtSnapshotRequest{Pkt: p}); err != nil && !errors.Is(err, io.EOF) {\n\treturn fmt.Errorf(\"failed to send data chunk: %w\", err)\n}\n// after\nif err := out.Send(&api.StreamExtSnapshotRequest{Pkt: p}); err != nil && !errors.Is(err, io.EOF) {\n\tif s, ok := status.FromError(err); ok && s.Code() == codes.Unavailable {\n\t\treturn fmt.Errorf(\"connection lost while sending chunk for group %d: %w\", groupId, err)\n\t}\n\treturn fmt.Errorf(\"failed to send data chunk: %w\", err)\n}","handlingStrategy":"retry","validationCode":"if ctx.Err() != nil {\n\treturn fmt.Errorf(\"cancelled before streaming; fix deadline or cancellation: %w\", ctx.Err())\n}","typeGuard":"func isTransportBreak(err error) bool {\n\ts, ok := status.FromError(err)\n\treturn ok && (s.Code() == codes.Unavailable || s.Code() == codes.Canceled || s.Code() == codes.Internal)\n}","tryCatchPattern":"if err := out.Send(&api.StreamExtSnapshotRequest{Pkt: p}); err != nil && !errors.Is(err, io.EOF) {\n\tif isTransportBreak(err) {\n\t\treturn errRetryWholeStream // stream not resumable; caller restarts group\n\t}\n\treturn fmt.Errorf(\"failed to send data chunk: %w\", err)\n}","preventionTips":["Enable gRPC keepalive on both ends of the connection.","Bypass or reconfigure LB idle timeouts for the import stream.","Pin imports to a stable Alpha (avoid pod eviction windows).","Test with a small pdir first to validate network stability."],"tags":["grpc","network","stream","snapshot-import"],"backgroundTag":"grpc-send-failed","analyzedSha":"759e242be62c91f8d084da06ad0c8d21256d9c07","analyzedAt":"2026-09-01T14:42:12.034Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}