{"record":{"id":"79bba9112384d77d","repo":"dgraph-io/dgraph","slug":"failed-to-establish-stream-with-leader-v","errorCode":null,"errorMessage":"failed to establish stream with leader: %v","messagePattern":"failed to establish stream with leader: (.+?)","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"worker/import.go","lineNumber":378,"sourceCode":"\tgroupId := req.GroupId\n\tif groupId == groups().Node.gid {\n\t\tglog.Infof(\"[import] streaming external snapshot to current group [%v]\", groupId)\n\t\treturn streamInGroup(stream, true)\n\t}\n\n\tglog.Infof(\"[import] streaming external snapshot to other group [%v]\", groupId)\n\tpl := groups().Leader(groupId)\n\tif pl == nil {\n\t\tglog.Errorf(\"[import] unable to connect to the leader of group [%v]\", groupId)\n\t\treturn fmt.Errorf(\"unable to connect to the leader of group [%v] : %v\", groupId, conn.ErrNoConnection)\n\t}\n\n\tcon := pl.Get()\n\tc := pb.NewWorkerClient(con)\n\talphaStream, err := c.StreamExtSnapshot(stream.Context())\n\tif err != nil {\n\t\tglog.Errorf(\"[import] failed to establish stream with leader: %v\", err)\n\t\treturn fmt.Errorf(\"failed to establish stream with leader: %v\", err)\n\t}\n\tglog.Infof(\"[import] [forward %d -> %d] start\", groups().Node.gid, groupId)\n\tglog.Infof(\"[import] [forward %v -> %d] start\", groups().Node.MyAddr, groups().Leader(groupId).Addr)\n\n\tglog.Infof(\"[import] sending forward true to leader of group [%v]\", groupId)\n\tforwardReq := &api.StreamExtSnapshotRequest{Forward: true}\n\tif err := alphaStream.Send(forwardReq); err != nil {\n\t\tglog.Errorf(\"[import] failed to send forward request: %v\", err)\n\t\treturn fmt.Errorf(\"failed to send forward request: %v\", err)\n\t}\n\n\treturn pipeTwoStream(stream, alphaStream, groupId)\n}\n\nfunc pipeTwoStream(in api.Dgraph_StreamExtSnapshotServer, out pb.Worker_StreamExtSnapshotClient, groupId uint32) error {\n\tcurrentGroup := groups().Node.gid\n\tctx := in.Context()\n","sourceCodeStart":360,"sourceCodeEnd":396,"githubUrl":"https://github.com/dgraph-io/dgraph/blob/759e242be62c91f8d084da06ad0c8d21256d9c07/worker/import.go#L360-L396","documentation":"After resolving the target group's leader, InStream opens the client side of the stream with c.StreamExtSnapshot(stream.Context()). If the gRPC call fails to start, the node logs and returns this error carrying the underlying transport/RPC error. It means a connection to the leader exists but the streaming RPC itself could not be established.","triggerScenarios":"c.StreamExtSnapshot(ctx) returns a non-nil err: the leader rejected the RPC (unimplemented/wrong version), the connection was torn down between Get() and the call, deadlines/context cancellation, or TLS/auth mismatch on the internal port.","commonSituations":"Mixed-version cluster where an old alpha does not implement StreamExtSnapshot; leader restarted between leader lookup and RPC; load balancer or proxy that does not support gRPC streaming; mTLS misconfiguration between alphas.","solutions":["Check the wrapped %v detail for the root cause (Unavailable, Unimplemented, deadline, TLS)","Ensure all alphas run the same Dgraph version that supports StreamExtSnapshot","Verify direct connectivity to the leader's internal address, bypassing any L7 proxy that breaks bidi streaming","Re-run the import; transient Unavailable after a leader change usually resolves once the new leader is warm"],"exampleFix":"// before\nalphaStream, err := c.StreamExtSnapshot(stream.Context())\nif err != nil {\n    return fmt.Errorf(\"failed to establish stream with leader: %v\", err)\n}\n// after: surface status code and add a short retry for transient Unavailable\nalphaStream, err := c.StreamExtSnapshot(stream.Context())\nif err != nil {\n    if st, ok := status.FromError(err); ok && st.Code() == codes.Unavailable {\n        time.Sleep(time.Second)\n        alphaStream, err = c.StreamExtSnapshot(stream.Context())\n    }\n    if err != nil {\n        return fmt.Errorf(\"failed to establish stream with leader: %w\", err)\n    }\n}","handlingStrategy":"retry","validationCode":"// ensure version compatibility and direct connectivity first\ndgraphVersionCheck(alphas) // all must implement StreamExtSnapshot\ntelnetOrDial(leaderAddr, internalPort) // RPC port reachable, no L7 proxy in path","typeGuard":"func canEstablishStream(c pb.WorkerClient, ctx context.Context) bool {\n    _, err := c.StreamExtSnapshot(ctx)\n    return err == nil || status.Code(err) != codes.Unimplemented\n}","tryCatchPattern":"if err != nil && strings.Contains(err.Error(), \"failed to establish stream with leader\") {\n    st, _ := status.FromError(errors.Unwrap(err))\n    switch st.Code() {\n    case codes.Unavailable: /* retry */\n    case codes.Unimplemented: /* upgrade cluster */\n    }\n}","preventionTips":["Keep all alphas on the same Dgraph version","Avoid HTTP/L7 proxies on internal gRPC ports","Set sane deadlines but allow generous time for snapshot RPCs","Check mTLS/TLS config parity between nodes"],"tags":["grpc","streaming","network","import"],"backgroundTag":"grpc-stream-establishment-failed","analyzedSha":"759e242be62c91f8d084da06ad0c8d21256d9c07","analyzedAt":"2026-09-01T14:42:12.034Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}