{"record":{"id":"3256a6fc74202df1","repo":"dgraph-io/dgraph","slug":"failed-to-run-in-group-streaming-v","errorCode":null,"errorMessage":"failed to run in group streaming: %v","messagePattern":"failed to run in group streaming: (.+?)","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"worker/import.go","lineNumber":626,"sourceCode":"\t\t\t})\n\t\t}\n\t}\n\n\teg.Go(func() error {\n\t\tdefer ps.close()\n\t\tdefer func() {\n\t\t\tif err := stream.Send(&api.StreamExtSnapshotResponse{}); err != nil {\n\t\t\t\tglog.Errorf(\"[import] failed to send close on in: %v\", err)\n\t\t\t}\n\t\t}()\n\t\tif err := ps.handlePublisher(errGCtx, stream); err != nil {\n\t\t\treturn fmt.Errorf(\"failed to run publisher: %v\", err)\n\t\t}\n\t\treturn nil\n\t})\n\n\tif err := eg.Wait(); err != nil {\n\t\treturn fmt.Errorf(\"failed to run in group streaming: %v\", err)\n\t}\n\t// Sends a StreamExtSnapshotResponse with the Finish flag set to true to indicate\n\t// the completion of the streaming process. If an error occurs during the send\n\t// operation, it is returned for further handling.\n\tif err := stream.Send(&api.StreamExtSnapshotResponse{Finish: true}); err != nil {\n\t\tglog.Errorf(\"failed to send done signal: %v\", err)\n\t}\n\n\t// If this node is the leader and fails to reach a majority of nodes, we return an error.\n\t// This ensures that the data is reliably received by enough nodes before proceeding.\n\tif forward && !checkMajority(successfulNodes) {\n\t\tglog.Error(\"[import] Majority of nodes failed to receive data.\")\n\t\treturn fmt.Errorf(\"failed to send data to majority of the nodes\")\n\t}\n\treturn nil\n}\n\nfunc updateNodeStatus(ps *sync.RWMutex, successfulNodes map[string]bool, addr string, status bool) {","sourceCodeStart":608,"sourceCodeEnd":644,"githubUrl":"https://github.com/dgraph-io/dgraph/blob/759e242be62c91f8d084da06ad0c8d21256d9c07/worker/import.go#L608-L644","documentation":"streamInGroup runs the publisher, local subscriber, and forwarders in an errgroup. If any goroutine returns an error (publisher failure, local subscriber failure such as a Badger write error, etc.), eg.Wait returns it and streamInGroup wraps it as 'failed to run in group streaming: %v'. This is the top-level wrapper for any intra-group streaming failure on this node.","triggerScenarios":"Any member goroutine of the errgroup fails: handlePublisher error, runLocalSubscriber error (e.g. Badger write failure, downstream ack error relayed through pipeTwoStream), or ctx cancellation of the stream.","commonSituations":"Disk full on the follower writing to Badger; one group member disconnected so the leader's forwarder aborts; client canceled the import; gRPC stream reset mid-import.","solutions":["Look earlier in the logs for '[import:flush] failed to run local subscriber' or 'failed to run forward subscriber' to find the root cause.","Fix the underlying cause (disk space, network to peers, client stability) and retry the whole import.","Verify all group members are alive and serving the group before starting the snapshot import.","Ensure Badger storage has sufficient free space and healthy disks."],"exampleFix":"null","handlingStrategy":"try-catch","validationCode":"// pre-flight: all group members alive, disk OK, quorum possible\nfor _, m := range members { if m.AmDead || !reachable(m.Addr) { return fmt.Errorf(\"member %s not ready\", m.Addr) } }","typeGuard":null,"tryCatchPattern":"err := streamSnapshot(ctx)\nif err != nil && strings.Contains(err.Error(), \"failed to run in group streaming\") {\n    // scan logs for local subscriber / forwarder root cause, fix, then full retry\n}","preventionTips":["Pre-flight check every group member before bulk import","Guarantee ample free disk on all followers (Badger writes fail hard)","Run imports during stable network windows; avoid concurrent maintenance"],"tags":["grpc","streaming","distributed-systems","import"],"backgroundTag":"grpc-stream-aborted","analyzedSha":"759e242be62c91f8d084da06ad0c8d21256d9c07","analyzedAt":"2026-09-01T14:42:12.034Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}