func (a *internalAPIServer) AddShard(shard uint64) error {
	version, ctx, err := a.versionAndCtx(context.Background())
	if err != nil {
		return err
	}
	if version == route.InvalidVersion {
		return nil
	}
	clientConn, err := a.router.GetMasterOrReplicaClientConn(shard, version)
	if err != nil {
		return err
	}
	repoInfos, err := pfs.NewInternalApiClient(clientConn).ListRepo(ctx, &pfs.ListRepoRequest{})
	if err != nil {
		return err
	}
	for _, repoInfo := range repoInfos.RepoInfo {
		if err := a.driver.CreateRepo(repoInfo.Repo); err != nil {
			return err
		}
		commitInfos, err := pfs.NewInternalApiClient(clientConn).ListCommit(ctx, &pfs.ListCommitRequest{Repo: repoInfo.Repo})
		if err != nil {
			return err
		}
		for i := range commitInfos.CommitInfo {
			commit := commitInfos.CommitInfo[len(commitInfos.CommitInfo)-(i+1)].Commit
			commitInfo, err := a.driver.InspectCommit(commit, map[uint64]bool{shard: true})
			if err != nil {
				return err
			}
			if commitInfo != nil {
				// we already have the commit so nothing to do
				continue
			}
			pullDiffRequest := &pfs.PullDiffRequest{
				Commit: commit,
				Shard:  shard,
			}
			pullDiffClient, err := pfs.NewInternalApiClient(clientConn).PullDiff(ctx, pullDiffRequest)
			if err != nil {
				return err
			}
			diffReader := protostream.NewStreamingBytesReader(pullDiffClient)
			a.driver.PushDiff(commit, diffReader)
		}
	}
	return nil
}

func RunTest(
	t *testing.T,
	f func(t *testing.T, apiClient pfs.ApiClient, internalAPIClient pfs.InternalApiClient),
) {
	discoveryClient, err := getEtcdClient()
	require.NoError(t, err)
	grpctest.Run(
		t,
		testNumServers,
		func(servers map[string]*grpc.Server) {
			registerFunc(getDriver(t), discoveryClient, servers)
		},
		func(t *testing.T, clientConns map[string]*grpc.ClientConn) {
			var clientConn *grpc.ClientConn
			for _, c := range clientConns {
				clientConn = c
				break
			}
			f(
				t,
				pfs.NewApiClient(
					clientConn,
				),
				pfs.NewInternalApiClient(
					clientConn,
				),
			)
		},
	)
}

func (a *apiServer) CreateRepo(ctx context.Context, request *pfs.CreateRepoRequest) (*google_protobuf.Empty, error) {
	version, ctx, err := a.versionAndCtx(ctx)
	if err != nil {
		return nil, err
	}
	clientConns, err := a.router.GetAllClientConns(version)
	if err != nil {
		return nil, err
	}
	for _, clientConn := range clientConns {
		if _, err := pfs.NewInternalApiClient(clientConn).CreateRepo(ctx, request); err != nil {
			return nil, err
		}
	}
	// Create the initial commit
	if _, err = a.StartCommit(ctx, &pfs.StartCommitRequest{
		Parent: nil,
		Commit: &pfs.Commit{
			Repo: request.Repo,
			Id:   InitialCommitID,
		},
	}); err != nil {
		return nil, err
	}
	if _, err = a.FinishCommit(ctx, &pfs.FinishCommitRequest{
		Commit: &pfs.Commit{
			Repo: request.Repo,
			Id:   InitialCommitID,
		},
	}); err != nil {
		return nil, err
	}
	return emptyInstance, nil
}

func (a *internalAPIServer) commitToReplicas(ctx context.Context, commit *pfs.Commit) error {
	version, err := a.getVersion(ctx)
	if err != nil {
		return err
	}
	shards, err := a.router.GetMasterShards(version)
	if err != nil {
		return err
	}
	for shard := range shards {
		clientConns, err := a.router.GetReplicaClientConns(shard, version)
		if err != nil {
			return err
		}
		var diff bytes.Buffer
		if err = a.driver.PullDiff(commit, shard, &diff); err != nil {
			return err
		}
		for _, clientConn := range clientConns {
			if _, err = pfs.NewInternalApiClient(clientConn).PushDiff(
				ctx,
				&pfs.PushDiffRequest{
					Commit: commit,
					Shard:  shard,
					Value:  diff.Bytes(),
				},
			); err != nil {
				return err
			}
		}
	}
	return nil
}

func RunTest(
	t *testing.T,
	f func(*testing.T, pfs.ApiClient, pfs.InternalApiClient, Cluster),
) {
	discoveryClient, err := getEtcdClient()
	require.NoError(t, err)
	var cluster Cluster
	prototest.RunT(
		t,
		testNumServers,
		func(servers map[string]*grpc.Server) {
			cluster = registerFunc(t, discoveryClient, servers)
		},
		func(t *testing.T, clientConns map[string]*grpc.ClientConn) {
			var clientConn *grpc.ClientConn
			for _, c := range clientConns {
				clientConn = c
				break
			}
			f(
				t,
				pfs.NewApiClient(
					clientConn,
				),
				pfs.NewInternalApiClient(
					clientConn,
				),
				cluster,
			)
		},
	)
	cluster.Shutdown()
}

func (a *combinedAPIServer) PushDiff(ctx context.Context, pushDiffRequest *pfs.PushDiffRequest) (*google_protobuf.Empty, error) {
	clientConn, err := a.getClientConnIfNecessary(int(pushDiffRequest.Shard), false)
	if err != nil {
		return nil, err
	}
	if clientConn != nil {
		return pfs.NewInternalApiClient(clientConn).PushDiff(ctx, pushDiffRequest)
	}
	return emptyInstance, a.driver.PushDiff(pushDiffRequest.Commit, bytes.NewReader(pushDiffRequest.Value))
}

func (a *apiServer) DeleteFile(ctx context.Context, request *pfs.DeleteFileRequest) (*google_protobuf.Empty, error) {
	version, ctx, err := a.versionAndCtx(ctx)
	if err != nil {
		return nil, err
	}
	clientConn, err := a.getClientConnForFile(request.File, version)
	if err != nil {
		return nil, err
	}
	return pfs.NewInternalApiClient(clientConn).DeleteFile(ctx, request)
}

func (a *apiServer) ListRepo(ctx context.Context, request *pfs.ListRepoRequest) (*pfs.RepoInfos, error) {
	version, ctx, err := a.versionAndCtx(ctx)
	if err != nil {
		return nil, err
	}
	clientConn, err := a.getClientConn(version)
	if err != nil {
		return nil, err
	}
	return pfs.NewInternalApiClient(clientConn).ListRepo(ctx, request)
}

func (a *apiServer) InspectFile(ctx context.Context, request *pfs.InspectFileRequest) (*pfs.FileInfo, error) {
	version, ctx, err := a.versionAndCtx(ctx)
	if err != nil {
		return nil, err
	}
	clientConn, err := a.getClientConnForFile(request.File, version)
	if err != nil {
		return nil, err
	}
	return pfs.NewInternalApiClient(clientConn).InspectFile(ctx, request)
}

func (a *apiServer) GetBlock(request *pfs.GetBlockRequest, apiGetBlockServer pfs.Api_GetBlockServer) (retErr error) {
	version, ctx, err := a.versionAndCtx(context.Background())
	if err != nil {
		return err
	}
	clientConn, err := a.getClientConnForBlock(request.Block, version)
	if err != nil {
		return err
	}
	blockGetClient, err := pfs.NewInternalApiClient(clientConn).GetBlock(ctx, request)
	if err != nil {
		return err
	}
	return protostream.RelayFromStreamingBytesClient(blockGetClient, apiGetBlockServer)
}

func (a *apiServer) PutBlock(ctx context.Context, request *pfs.PutBlockRequest) (*pfs.Block, error) {
	version, ctx, err := a.versionAndCtx(ctx)
	if err != nil {
		return nil, err
	}
	block := a.sharder.GetBlock(request.Value)
	clientConn, err := a.getClientConnForBlock(block, version)
	if err != nil {
		return nil, err
	}
	if _, err := pfs.NewInternalApiClient(clientConn).PutBlock(ctx, request); err != nil {
		return nil, err
	}
	return block, nil
}

func (a *apiServer) GetFile(request *pfs.GetFileRequest, apiGetFileServer pfs.Api_GetFileServer) error {
	version, ctx, err := a.versionAndCtx(context.Background())
	if err != nil {
		return err
	}
	clientConn, err := a.getClientConnForFile(request.File, version)
	if err != nil {
		return err
	}
	fileGetClient, err := pfs.NewInternalApiClient(clientConn).GetFile(ctx, request)
	if err != nil {
		return err
	}
	return protostream.RelayFromStreamingBytesClient(fileGetClient, apiGetFileServer)
}

func (a *combinedAPIServer) Master(shard int) error {
	clientConns, err := a.router.GetReplicaClientConns(shard)
	if err != nil {
		return err
	}
	for _, clientConn := range clientConns {
		apiClient := pfs.NewApiClient(clientConn)
		response, err := apiClient.ListRepositories(context.Background(), &pfs.ListRepositoriesRequest{})
		if err != nil {
			return err
		}
		for _, repository := range response.Repository {
			if err := a.driver.InitRepository(repository, map[int]bool{shard: true}); err != nil {
				return err
			}
			response, err := apiClient.ListCommits(context.Background(), &pfs.ListCommitsRequest{Repository: repository})
			if err != nil {
				return err
			}
			localCommitInfo, err := a.driver.ListCommits(repository, shard)
			if err != nil {
				return err
			}
			for i, commitInfo := range response.CommitInfo {
				if i < len(localCommitInfo) {
					if *commitInfo != *localCommitInfo[i] {
						return fmt.Errorf("divergent data")
					}
					continue
				}
				pullDiffClient, err := pfs.NewInternalApiClient(clientConn).PullDiff(
					context.Background(),
					&pfs.PullDiffRequest{
						Commit: commitInfo.Commit,
						Shard:  uint64(shard),
					},
				)
				if err != nil {
					return err
				}
				if err := a.driver.PushDiff(commitInfo.Commit, protostream.NewStreamingBytesReader(pullDiffClient)); err != nil {
					return err
				}
			}
		}
	}
	return nil
}

func (a *apiServer) FinishCommit(ctx context.Context, request *pfs.FinishCommitRequest) (*google_protobuf.Empty, error) {
	version, ctx, err := a.versionAndCtx(ctx)
	if err != nil {
		return nil, err
	}
	clientConns, err := a.router.GetAllClientConns(version)
	if err != nil {
		return nil, err
	}
	for _, clientConn := range clientConns {
		if _, err := pfs.NewInternalApiClient(clientConn).FinishCommit(ctx, request); err != nil {
			return nil, err
		}
	}
	return emptyInstance, nil
}

func (a *apiServer) ListFile(ctx context.Context, request *pfs.ListFileRequest) (*pfs.FileInfos, error) {
	version, ctx, err := a.versionAndCtx(ctx)
	if err != nil {
		return nil, err
	}
	clientConns, err := a.router.GetAllClientConns(version)
	if err != nil {
		return nil, err
	}
	var wg sync.WaitGroup
	var lock sync.Mutex
	var fileInfos []*pfs.FileInfo
	seenDirectories := make(map[string]bool)
	var loopErr error
	for _, clientConn := range clientConns {
		wg.Add(1)
		go func(clientConn *grpc.ClientConn) {
			defer wg.Done()
			subFileInfos, err := pfs.NewInternalApiClient(clientConn).ListFile(ctx, request)
			lock.Lock()
			defer lock.Unlock()
			if err != nil {
				if loopErr == nil {
					loopErr = err
				}
				return
			}
			for _, fileInfo := range subFileInfos.FileInfo {
				if fileInfo.FileType == pfs.FileType_FILE_TYPE_DIR {
					if seenDirectories[fileInfo.File.Path] {
						continue
					}
					seenDirectories[fileInfo.File.Path] = true
				}
				fileInfos = append(fileInfos, fileInfo)
			}
		}(clientConn)
	}
	wg.Wait()
	if loopErr != nil {
		return nil, loopErr
	}
	return &pfs.FileInfos{
		FileInfo: fileInfos,
	}, nil
}

func (a *combinedAPIServer) PullDiff(pullDiffRequest *pfs.PullDiffRequest, apiPullDiffServer pfs.InternalApi_PullDiffServer) error {
	clientConn, err := a.getClientConnIfNecessary(int(pullDiffRequest.Shard), false)
	if err != nil {
		return err
	}
	if clientConn != nil {
		apiPullDiffClient, err := pfs.NewInternalApiClient(clientConn).PullDiff(context.Background(), pullDiffRequest)
		if err != nil {
			return err
		}
		return protoutil.RelayFromStreamingBytesClient(apiPullDiffClient, apiPullDiffServer)
	}
	var buffer bytes.Buffer
	a.driver.PullDiff(pullDiffRequest.Commit, int(pullDiffRequest.Shard), &buffer)
	return protoutil.WriteToStreamingBytesServer(
		&buffer,
		apiPullDiffServer,
	)
}

func (a *apiServer) StartCommit(ctx context.Context, request *pfs.StartCommitRequest) (*pfs.Commit, error) {
	version, ctx, err := a.versionAndCtx(ctx)
	if err != nil {
		return nil, err
	}
	clientConns, err := a.router.GetAllClientConns(version)
	if err != nil {
		return nil, err
	}
	if request.Commit == nil {
		request.Commit = &pfs.Commit{
			Repo: request.Parent.Repo,
			Id:   strings.Replace(uuid.NewV4().String(), "-", "", -1),
		}
	}
	for _, clientConn := range clientConns {
		if _, err := pfs.NewInternalApiClient(clientConn).StartCommit(ctx, request); err != nil {
			return nil, err
		}
	}
	return request.Commit, nil
}

func (a *apiServer) ListChange(ctx context.Context, request *pfs.ListChangeRequest) (*pfs.Changes, error) {
	version, ctx, err := a.versionAndCtx(ctx)
	if err != nil {
		return nil, err
	}
	clientConns, err := a.router.GetAllClientConns(version)
	if err != nil {
		return nil, err
	}
	var wg sync.WaitGroup
	var lock sync.Mutex
	var changes []*pfs.Change
	var loopErr error
	for _, clientConn := range clientConns {
		wg.Add(1)
		go func(clientConn *grpc.ClientConn) {
			defer wg.Done()
			subChanges, err := pfs.NewInternalApiClient(clientConn).ListChange(ctx, request)
			lock.Lock()
			defer lock.Unlock()
			if err != nil {
				if loopErr == nil {
					loopErr = err
				}
				return
			}
			for _, change := range subChanges.Change {
				changes = append(changes, change)
			}
		}(clientConn)
	}
	wg.Wait()
	if loopErr != nil {
		return nil, loopErr
	}
	return &pfs.Changes{
		Change: changes,
	}, nil
}

func (a *apiServer) ListCommit(ctx context.Context, request *pfs.ListCommitRequest) (*pfs.CommitInfos, error) {
	version, ctx, err := a.versionAndCtx(ctx)
	if err != nil {
		return nil, err
	}
	clientConns, err := a.router.GetAllClientConns(version)
	if err != nil {
		return nil, err
	}
	var wg sync.WaitGroup
	var lock sync.Mutex
	var commitInfos [][]*pfs.CommitInfo
	var loopErr error
	for _, clientConn := range clientConns {
		wg.Add(1)
		go func(clientConn *grpc.ClientConn) {
			defer wg.Done()
			subCommitInfos, err := pfs.NewInternalApiClient(clientConn).ListCommit(ctx, request)
			lock.Lock()
			defer lock.Unlock()
			if err != nil {
				loopErr = err
				return
			}
			commitInfos = append(commitInfos, subCommitInfos.CommitInfo)
		}(clientConn)
	}
	wg.Wait()
	if loopErr != nil {
		return nil, loopErr
	}
	idToCommitInfo := make(map[string]*pfs.CommitInfo)
	idToSeenCount := make(map[string]int)
	for _, subCommitInfos := range commitInfos {
		for _, subCommitInfo := range subCommitInfos {
			idToSeenCount[subCommitInfo.Commit.Id] = idToSeenCount[subCommitInfo.Commit.Id] + 1
			commitInfo, ok := idToCommitInfo[subCommitInfo.Commit.Id]
			if !ok {
				idToCommitInfo[subCommitInfo.Commit.Id] = subCommitInfo
				continue
			}
			if commitInfo.CommitType != pfs.CommitType_COMMIT_TYPE_READ {
				commitInfo.CommitType = subCommitInfo.CommitType
			}
			if prototime.TimestampToTime(subCommitInfo.Started).Before(prototime.TimestampToTime(commitInfo.Started)) {
				commitInfo.Started = subCommitInfo.Started
			}
			if commitInfo.Finished != nil {
				if subCommitInfo.Finished == nil {
					commitInfo.Finished = nil
				} else if prototime.TimestampToTime(subCommitInfo.Finished).After(prototime.TimestampToTime(commitInfo.Finished)) {
					commitInfo.Finished = subCommitInfo.Finished
				}
			}
			commitInfo.CommitBytes += subCommitInfo.CommitBytes
			commitInfo.TotalBytes += subCommitInfo.TotalBytes
		}
	}
	var result []*pfs.CommitInfo
	for _, subCommitInfos := range commitInfos {
		for _, commitInfo := range subCommitInfos {
			if idToSeenCount[commitInfo.Commit.Id] == len(commitInfos) {
				result = append(result, idToCommitInfo[commitInfo.Commit.Id])
			}
		}
		break
	}
	return &pfs.CommitInfos{CommitInfo: result}, nil
}

func (a *apiServer) PutFile(putFileServer pfs.Api_PutFileServer) (retErr error) {
	version, ctx, err := a.versionAndCtx(putFileServer.Context())
	if err != nil {
		return err
	}
	defer func() {
		if err := putFileServer.SendAndClose(emptyInstance); err != nil && retErr == nil {
			retErr = err
		}
	}()
	request, err := putFileServer.Recv()
	if err != nil {
		return err
	}
	if strings.HasPrefix(request.File.Path, "/") {
		// This is a subtle error case, the paths foo and /foo will hash to
		// different shards but will produce the same change once they get to
		// those shards due to how path.Join. This can go wrong in a number of
		// ways so we forbid leading slashes.
		return fmt.Errorf("pachyderm: leading slash in path: %s", request.File.Path)
	}
	if request.FileType == pfs.FileType_FILE_TYPE_DIR {
		if len(request.Value) > 0 {
			return fmt.Errorf("PutFileRequest shouldn't have type dir and a value")
		}
		clientConns, err := a.router.GetAllClientConns(version)
		if err != nil {
			return err
		}
		for _, clientConn := range clientConns {
			putFileClient, err := pfs.NewInternalApiClient(clientConn).PutFile(ctx)
			if err != nil {
				return err
			}
			if err := putFileClient.Send(request); err != nil {
				return err
			}
			if _, err := putFileClient.CloseAndRecv(); err != nil {
				return err
			}
		}
		return nil
	}
	clientConn, err := a.getClientConnForFile(request.File, version)
	if err != nil {
		return err
	}
	putFileClient, err := pfs.NewInternalApiClient(clientConn).PutFile(ctx)
	if err != nil {
		return err
	}
	defer func() {
		if _, err := putFileClient.CloseAndRecv(); err != nil && retErr == nil {
			retErr = err
		}
	}()
	if err := putFileClient.Send(request); err != nil {
		return err
	}
	for {
		request, err := putFileServer.Recv()
		if err != nil {
			if err == io.EOF {
				break
			}
			return err
		}
		if err := putFileClient.Send(request); err != nil {
			return err
		}
	}
	if _, err := putFileClient.CloseAndRecv(); err != nil {
		return err
	}
	return nil
}