dgraph-io/dgraph · error

failed to establish stream with leader: %v

Error message

failed to establish stream with leader: %v

What it means

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.

Source

Thrown at worker/import.go:378

	groupId := req.GroupId
	if groupId == groups().Node.gid {
		glog.Infof("[import] streaming external snapshot to current group [%v]", groupId)
		return streamInGroup(stream, true)
	}

	glog.Infof("[import] streaming external snapshot to other group [%v]", groupId)
	pl := groups().Leader(groupId)
	if pl == nil {
		glog.Errorf("[import] unable to connect to the leader of group [%v]", groupId)
		return fmt.Errorf("unable to connect to the leader of group [%v] : %v", groupId, conn.ErrNoConnection)
	}

	con := pl.Get()
	c := pb.NewWorkerClient(con)
	alphaStream, err := c.StreamExtSnapshot(stream.Context())
	if err != nil {
		glog.Errorf("[import] failed to establish stream with leader: %v", err)
		return fmt.Errorf("failed to establish stream with leader: %v", err)
	}
	glog.Infof("[import] [forward %d -> %d] start", groups().Node.gid, groupId)
	glog.Infof("[import] [forward %v -> %d] start", groups().Node.MyAddr, groups().Leader(groupId).Addr)

	glog.Infof("[import] sending forward true to leader of group [%v]", groupId)
	forwardReq := &api.StreamExtSnapshotRequest{Forward: true}
	if err := alphaStream.Send(forwardReq); err != nil {
		glog.Errorf("[import] failed to send forward request: %v", err)
		return fmt.Errorf("failed to send forward request: %v", err)
	}

	return pipeTwoStream(stream, alphaStream, groupId)
}

func pipeTwoStream(in api.Dgraph_StreamExtSnapshotServer, out pb.Worker_StreamExtSnapshotClient, groupId uint32) error {
	currentGroup := groups().Node.gid
	ctx := in.Context()

View on GitHub (pinned to 759e242be6)

Solutions

1. Check the wrapped %v detail for the root cause (Unavailable, Unimplemented, deadline, TLS)
2. Ensure all alphas run the same Dgraph version that supports StreamExtSnapshot
3. Verify direct connectivity to the leader's internal address, bypassing any L7 proxy that breaks bidi streaming
4. Re-run the import; transient Unavailable after a leader change usually resolves once the new leader is warm

Example fix

// before
alphaStream, err := c.StreamExtSnapshot(stream.Context())
if err != nil {
    return fmt.Errorf("failed to establish stream with leader: %v", err)
}
// after: surface status code and add a short retry for transient Unavailable
alphaStream, err := c.StreamExtSnapshot(stream.Context())
if err != nil {
    if st, ok := status.FromError(err); ok && st.Code() == codes.Unavailable {
        time.Sleep(time.Second)
        alphaStream, err = c.StreamExtSnapshot(stream.Context())
    }
    if err != nil {
        return fmt.Errorf("failed to establish stream with leader: %w", err)
    }
}

// ensure version compatibility and direct connectivity first
dgraphVersionCheck(alphas) // all must implement StreamExtSnapshot
telnetOrDial(leaderAddr, internalPort) // RPC port reachable, no L7 proxy in path

func canEstablishStream(c pb.WorkerClient, ctx context.Context) bool {
    _, err := c.StreamExtSnapshot(ctx)
    return err == nil || status.Code(err) != codes.Unimplemented
}

if err != nil && strings.Contains(err.Error(), "failed to establish stream with leader") {
    st, _ := status.FromError(errors.Unwrap(err))
    switch st.Code() {
    case codes.Unavailable: /* retry */
    case codes.Unimplemented: /* upgrade cluster */
    }
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• failed to connect to endpoint [%s]: %w
• failed to send forward request: %v
• recv final downstream(%d): %w
• ack data downstream(%d): %w
• send ack upstream: %w

AI-assisted analysis of dgraph-io/dgraph@759e242be6 (2026-09-01). Data as JSON: /api/errors/79bba9112384d77d. Report an issue: GitHub.