{"record":{"id":"10d357e574c99d71","repo":"dgraph-io/dgraph","slug":"failed-to-run-publisher-v","errorCode":null,"errorMessage":"failed to run publisher: %v","messagePattern":"failed to run publisher: (.+?)","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"worker/import.go","lineNumber":620,"sourceCode":"\t\t\t\t\treturn nil\n\t\t\t\t}\n\n\t\t\t\tupdateNodeStatus(&ps.RWMutex, successfulNodes, member.Addr, true)\n\t\t\t\tglog.Infof(\"[import] Successfully connected and streamed data to node: %v\", member.Addr)\n\t\t\t\treturn nil\n\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.\")","sourceCodeStart":602,"sourceCodeEnd":638,"githubUrl":"https://github.com/dgraph-io/dgraph/blob/759e242be62c91f8d084da06ad0c8d21256d9c07/worker/import.go#L602-L638","documentation":"In streamInGroup, the publisher goroutine reads from the incoming StreamExtSnapshot stream and fans messages out to subscribers (local writer and forwarders). If ps.handlePublisher returns an error (e.g. failed Recv on the incoming stream other than EOF), it is wrapped as 'failed to run publisher: %v'.","triggerScenarios":"The client streaming into StreamExtSnapshot breaks mid-stream: Recv returns a transport error or the context is canceled; malformed/failed receive inside the publisher loop.","commonSituations":"dgraph live loader / client disconnected during bulk import; network drop between proxy alpha and this leader; stream canceled because another group member (subscriber/forwarder) failed first via errgroup context.","solutions":["Check the wrapped cause for the real transport error and fix connectivity between the importing client and this alpha.","Retry the snapshot streaming from the beginning; errgroup aborts the whole stream.","Check alpha logs for the paired subscriber/forwarder errors that may have canceled the shared errgroup context.","Ensure the importing client is not timing out before the stream completes."],"exampleFix":"null","handlingStrategy":"try-catch","validationCode":"// verify client stability and connectivity before streaming\nif err := pingAlpha(proxyAddr); err != nil { return err }","typeGuard":null,"tryCatchPattern":"err := streamInGroupViaClient(ctx)\nif err != nil && strings.Contains(err.Error(), \"failed to run publisher\") {\n    root := errors.Unwrap(err)\n    // root is the transport/recv error; retry whole stream after fixing it\n}","preventionTips":["Keep the importing client process alive and connected for the full stream","Avoid network interruptions between the client/proxy alpha and the group leader","Check alpha logs for the first underlying error; this message is only a wrapper"],"tags":["grpc","streaming","pubsub","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"}