go/vcsinfo/bt_vcs/bt_vcs.go - buildbot - Git at Google

 package bt_vcs

 import (
 	"bytes"
 	"context"
 	"errors"
 	"math"
 	"sort"
 	"sync"
 	"time"

 	"go.skia.org/infra/go/depot_tools"
 	"go.skia.org/infra/go/eventbus"
 	"go.skia.org/infra/go/gitiles"
 	"go.skia.org/infra/go/gitstore"
 	"go.skia.org/infra/go/gitstore/bt_gitstore"
 	"go.skia.org/infra/go/skerr"
 	"go.skia.org/infra/go/sklog"
 	"go.skia.org/infra/go/util"
 	"go.skia.org/infra/go/vcsinfo"
 	"golang.org/x/sync/errgroup"
 )

 const (
 	// EV_NEW_GIT_COMMIT is the event that is fired when a previously unseen Git commit is available.
 	// The event data for this commit is of type []*vcsinfo.IndexCommit containing all new commits
 	// that have been added since the last commit was sent.
 	EV_NEW_GIT_COMMIT = "gitstore:new-git-commit"

 	// defaultWatchInterval is the interval at which we check for new commits being added to the repo.
 	defaultWatchInterval = time.Second * 10

 	// nCommits is a parameter for StartTracking to tune how many
 	// commits gets sent over the event bus when new commits are
 	// detected.
 	nCommits = 20
 )

 // BigTableVCS implements the vcsinfo.VCS interface based on a BT-backed GitStore.
 type BigTableVCS struct {
 	gitStore           gitstore.GitStore
 	repo               *gitiles.Repo
 	defaultBranch      string
 	secondaryVCS       vcsinfo.VCS
 	secondaryExtractor depot_tools.DEPSExtractor

 	branchInfo *gitstore.BranchPointer

 	// This mutex protects detailsCache and indexCommits
 	mutex sync.RWMutex
 	// detailsCache is for LongCommits so we don't have to query gitStore every time
 	detailsCache map[string]*vcsinfo.LongCommit
 	indexCommits []*vcsinfo.IndexCommit
 }

 // NewVCS returns an instance of vcsinfo.VCS that is backed by the given GitStore and uses the
 // gittiles.Repo to retrieve files. Each instance provides an interface to one branch.
 // If defaultBranch is gitstore.ALL_BRANCHES all commits in the repository are considered.
 // The instances of gitiles.Repo is only used to fetch files.
 func New(ctx context.Context, gitStore gitstore.GitStore, defaultBranch string, repo *gitiles.Repo) (*BigTableVCS, error) {
 	if gitStore == nil {
 		return nil, errors.New("Cannot have nil gitStore")
 	}
 	ret := &BigTableVCS{
 		gitStore:      gitStore,
 		repo:          repo,
 		defaultBranch: defaultBranch,
 		detailsCache:  map[string]*vcsinfo.LongCommit{},
 	}
 	if err := ret.Update(ctx, true, false); err != nil {
 		return nil, skerr.Wrapf(err, "could not perform initial update")
 	}

 	return ret, nil
 }

 // GetBranch implements the vcsinfo.VCS interface.
 func (b *BigTableVCS) GetBranch() string {
 	return b.defaultBranch
 }

 // SetSecondaryRepo allows to add a secondary repository and extractor to this instance.
 // It is not included in the constructor since it is currently only used by the Gold ingesters.
 func (b *BigTableVCS) SetSecondaryRepo(secVCS vcsinfo.VCS, extractor depot_tools.DEPSExtractor) {
 	b.secondaryVCS = secVCS
 	b.secondaryExtractor = extractor
 }

 // Update implements the vcsinfo.VCS interface
 func (b *BigTableVCS) Update(ctx context.Context, pull, allBranches bool) error {
 	// Check if we need to pull across all branches.
 	targetBranch := b.defaultBranch
 	if allBranches {
 		targetBranch = gitstore.ALL_BRANCHES
 	}

 	// Simulate a pull by fetching the latest head of the target branch.
 	if pull {
 		branchHeads, err := b.gitStore.GetBranches(ctx)
 		if err != nil {
 			return skerr.Wrap(err)
 		}

 		var ok bool
 		b.branchInfo, ok = branchHeads[targetBranch]
 		if !ok {
 			r := (b.gitStore.(*bt_gitstore.BigTableGitStore)).RepoURL
 			return skerr.Fmt("unable to find branch %q in BigTable repo %s", targetBranch, r)
 		}
 	}

 	// Get all index commits for the current branch.
 	if err := b.fetchIndexRange(ctx, 0, b.branchInfo.Index+1); err != nil {
 		return skerr.Wrapf(err, "could not fetch index commits [0,%d]", b.branchInfo.Index+1)
 	}
 	// warm the cache of long commits
 	hashes := func() []string {
 		b.mutex.RLock()
 		defer b.mutex.RUnlock()
 		hashes := make([]string, 0, len(b.indexCommits))
 		for _, ic := range b.indexCommits {
 			hashes = append(hashes, ic.Hash)
 		}
 		return hashes
 	}()

 	_, err := b.DetailsMulti(ctx, hashes, false)
 	return err
 }

 // From implements the vcsinfo.VCS interface
 func (b *BigTableVCS) From(start time.Time) []string {
 	b.mutex.RLock()
 	defer b.mutex.RUnlock()

 	// Add a millisecond because we only want commits after the startTime. Timestamps in git are
 	// only at second level granularity.
 	found := b.timeRange(start.Add(time.Millisecond), vcsinfo.MaxTime)
 	ret := make([]string, len(found))
 	for i, c := range found {
 		ret[i] = c.Hash
 	}
 	return ret
 }

 // Details implements the vcsinfo.VCS interface
 func (b *BigTableVCS) Details(ctx context.Context, hash string, includeBranchInfo bool) (*vcsinfo.LongCommit, error) {
 	// Check the cache first, being sure to unlock the mutex after so that
 	// if we have to poll BT for it, we aren't blocking the whole time.
 	rv := func() *vcsinfo.LongCommit {
 		b.mutex.RLock()
 		defer b.mutex.RUnlock()
 		c, ok := b.detailsCache[hash]
 		if ok {
 			// If the user is querying for branch info,
 			// we have to check to see if the cached value has it.
 			// If the cached value has it and the user didn't request it,
 			// return it anyway.
 			if !includeBranchInfo || len(c.Branches) > 0 {
 				return c
 			}
 		}
 		return nil
 	}()
 	if rv != nil {
 		return rv, nil
 	}

 	// cache miss, so query for it
 	c, err := b.details(ctx, hash, includeBranchInfo)
 	if err != nil {
 		return nil, skerr.Wrapf(err, "could not fetch Details for commit %s", hash)
 	}
 	if c != nil {
 		b.mutex.Lock()
 		defer b.mutex.Unlock()
 		b.detailsCache[hash] = c
 	}
 	return c, err
 }

 // DetailsMulti implements the vcsinfo.VCS interface
 func (b *BigTableVCS) DetailsMulti(ctx context.Context, hashes []string, includeBranchInfo bool) ([]*vcsinfo.LongCommit, error) {
 	// Instantiate a list of N nil values so we can directly insert them into the proper index.
 	rv := make([]*vcsinfo.LongCommit, len(hashes))

 	missedHashes := []string{}
 	// Index into hashes of which hashes were not in the cache
 	missedHashesIdx := []int{}

 	func() {
 		b.mutex.RLock()
 		defer b.mutex.RUnlock()
 		for i, hash := range hashes {
 			c, ok := b.detailsCache[hash]
 			// If the user is querying for branch info,
 			// we have to check to see if the cached value has it.
 			if ok && (!includeBranchInfo || len(c.Branches) > 0) {
 				rv[i] = c
 			} else {
 				missedHashesIdx = append(missedHashesIdx, i)
 				missedHashes = append(missedHashes, hash)
 			}
 		}
 	}()
 	if len(missedHashes) == 0 {
 		return rv, nil
 	}

 	// bulk fetch the missedHashes.  A simpler, but potentially slower, approach
 	// could just do a b.details() for each commit (parallelized with an errgroup)
 	commits, err := b.gitStore.Get(ctx, missedHashes)
 	if err != nil {
 		return nil, skerr.Wrapf(err, "Get missed hashes %q (superset %q)", missedHashes, hashes)
 	}

 	if includeBranchInfo {
 		branchPointers, err := b.gitStore.GetBranches(ctx)
 		if err != nil {
 			return nil, skerr.Wrap(err)
 		}

 		// We need to do an individual request per commit for its branch info
 		var egroup errgroup.Group
 		for _, c := range commits {
 			if c != nil {
 				// Create a closure since we pass each value of 'c' to its own go-routine.
 				func(c *vcsinfo.LongCommit) {
 					egroup.Go(func() error {
 						branches, err := b.getBranchInfo(ctx, c, branchPointers)
 						if err != nil {
 							return skerr.Wrapf(err, "getBranchInfo for commit %s", c.Hash)
 						}
 						c.Branches = branches
 						return nil
 					})
 				}(c)
 			}
 		}
 		if err := egroup.Wait(); err != nil {
 			return nil, skerr.Wrapf(err, "Fetching branch info for %q", missedHashes)
 		}
 	}

 	b.mutex.Lock()
 	defer b.mutex.Unlock()
 	for i, hash := range missedHashes {
 		c := commits[i]
 		if c != nil {
 			rv[missedHashesIdx[i]] = c
 			b.detailsCache[hash] = c
 		}
 	}

 	return rv, nil
 }

 // details returns all meta data details we care about.
 func (b *BigTableVCS) details(ctx context.Context, hash string, includeBranchInfo bool) (*vcsinfo.LongCommit, error) {
 	commits, err := b.gitStore.Get(ctx, []string{hash})
 	if err != nil {
 		return nil, err
 	}

 	if len(commits) == 0 || commits[0] == nil {
 		return nil, skerr.Fmt("Commit %s not found", hash)
 	}

 	if includeBranchInfo {
 		branchPointers, err := b.gitStore.GetBranches(ctx)
 		if err != nil {
 			return nil, skerr.Wrap(err)
 		}

 		branches, err := b.getBranchInfo(ctx, commits[0], branchPointers)
 		if err != nil {
 			return nil, skerr.Wrapf(err, "getting branch info for commit %s", commits[0].Hash)
 		}
 		commits[0].Branches = branches
 	}
 	return commits[0], nil
 }

 // getBranchInfo determines which branches contain the given commit 'c'.
 // This function can potentially spawn a huge number of goroutines (one per branch).
 func (b *BigTableVCS) getBranchInfo(ctx context.Context, c *vcsinfo.LongCommit, allBranches map[string]*gitstore.BranchPointer) (map[string]bool, error) {
 	ret := make(map[string]bool, len(allBranches))
 	var mutex sync.Mutex
 	var egroup errgroup.Group
 	for branchName := range allBranches {
 		if branchName != gitstore.ALL_BRANCHES {
 			func(branchName string) {
 				egroup.Go(func() error {
 					// Since we cannot look up a commit in a branch directly we query for all commits that
 					// occurred at that specific timestamp (Git has second granularity) on the target branch.
 					// Then we check whether the target commit is returned as part of the result.
 					commits, err := b.gitStore.RangeByTime(ctx, c.Timestamp, c.Timestamp.Add(time.Second), branchName)
 					if err != nil {
 						return skerr.Wrapf(err, "range query for branch %s", branchName)
 					}

 					// Iterate over the commits at the given timestamp. Most of the time there should
 					// only be one commit at a given one second time range.
 					for _, idxCommit := range commits {
 						if idxCommit.Hash == c.Hash {
 							mutex.Lock()
 							ret[branchName] = true
 							mutex.Unlock()
 							break
 						}
 					}
 					return nil
 				})
 			}(branchName)
 		}
 	}
 	if err := egroup.Wait(); err != nil {
 		return nil, skerr.Wrapf(err, "retrieving branch info for %s", c.Hash)
 	}
 	return ret, nil
 }

 // LastNIndex implements the vcsinfo.VCS interface
 func (b *BigTableVCS) LastNIndex(N int) []*vcsinfo.IndexCommit {
 	b.mutex.RLock()
 	defer b.mutex.RUnlock()

 	if N > len(b.indexCommits) {
 		N = len(b.indexCommits)
 	}
 	ret := make([]*vcsinfo.IndexCommit, 0, N)
 	return append(ret, b.indexCommits[len(b.indexCommits)-N:]...)
 }

 // Range implements the vcsinfo.VCS interface
 func (b *BigTableVCS) Range(begin, end time.Time) []*vcsinfo.IndexCommit {
 	return b.timeRange(begin, end)
 }

 // IndexOf implements the vcsinfo.VCS interface
 func (b *BigTableVCS) IndexOf(ctx context.Context, hash string) (int, error) {
 	b.mutex.RLock()
 	defer b.mutex.RUnlock()

 	for i := len(b.indexCommits) - 1; i >= 0; i-- {
 		if hash == b.indexCommits[i].Hash {
 			return b.indexCommits[i].Index, nil
 		}
 	}

 	return -1, skerr.Fmt("commit %s not found", hash)
 }

 // ByIndex implements the vcsinfo.VCS interface
 func (b *BigTableVCS) ByIndex(ctx context.Context, N int) (*vcsinfo.LongCommit, error) {
 	// findFn returns the hash when N is within commits
 	findFn := func(commits []*vcsinfo.IndexCommit) string {
 		i := sort.Search(len(commits), func(i int) bool { return commits[i].Index >= N })
 		return commits[i].Hash
 	}

 	hash := func() string {
 		b.mutex.RLock()
 		defer b.mutex.RUnlock()
 		if len(b.indexCommits) > 0 {
 			firstIdx := b.indexCommits[0].Index
 			lastIdx := b.indexCommits[len(b.indexCommits)-1].Index
 			if (N >= firstIdx) && (N <= lastIdx) {
 				return findFn(b.indexCommits)
 			}
 		}
 		return ""
 	}()

 	// Fetch the hash
 	if hash == "" {
 		return nil, skerr.Fmt("Hash index not found: %d", N)
 	}
 	return b.Details(ctx, hash, false)
 }

 // GetFile implements the vcsinfo.VCS interface
 func (b *BigTableVCS) GetFile(ctx context.Context, fileName, commitHash string) (string, error) {
 	var buf bytes.Buffer
 	if err := b.repo.ReadFileAtRef(ctx, fileName, commitHash, &buf); err != nil {
 		return "", skerr.Wrapf(err, "reading file %s @ %s via gitiles", fileName, commitHash)
 	}
 	return buf.String(), nil
 }

 // ResolveCommit implements the vcsinfo.VCS interface
 func (b *BigTableVCS) ResolveCommit(ctx context.Context, commitHash string) (string, error) {
 	if b.secondaryVCS == nil {
 		return "", vcsinfo.NoSecondaryRepo
 	}

 	foundCommit, err := b.secondaryExtractor.ExtractCommit(b.secondaryVCS.GetFile(ctx, "DEPS", commitHash))
 	if err != nil {
 		return "", err
 	}
 	return foundCommit, nil
 }

 // GetGitStore implements the gitstore.GitStoreBased interface
 func (b *BigTableVCS) GetGitStore() gitstore.GitStore {
 	return b.gitStore
 }

 // fetchIndexRange gets in the range [startIndex, endIndex).
 func (b *BigTableVCS) fetchIndexRange(ctx context.Context, startIndex, endIndex int) error {
 	newIC, err := b.gitStore.RangeN(ctx, startIndex, endIndex, b.defaultBranch)
 	if err != nil {
 		return err
 	}

 	if len(newIC) == 0 {
 		return nil
 	}

 	b.mutex.Lock()
 	defer b.mutex.Unlock()
 	b.indexCommits = newIC
 	return nil
 }

 func (b *BigTableVCS) timeRange(start time.Time, end time.Time) []*vcsinfo.IndexCommit {
 	n := len(b.indexCommits)
 	startIdx := 0
 	for ; startIdx < n; startIdx++ {
 		exp := b.indexCommits[startIdx].Timestamp.After(start) || b.indexCommits[startIdx].Timestamp.Equal(start)
 		if exp {
 			break
 		}
 	}

 	endIdx := startIdx
 	for ; endIdx < n; endIdx++ {
 		exp := b.indexCommits[endIdx].Timestamp.After(end) || b.indexCommits[endIdx].Timestamp.Equal(end)
 		if exp {
 			break
 		}
 	}

 	if endIdx <= startIdx {
 		return []*vcsinfo.IndexCommit{}
 	}
 	return b.indexCommits[startIdx:endIdx]
 }

 // StartTracking begins watching the repo for changes on a background thread.
 // When a new commit is detected a EV_NEW_GIT_COMMIT event is triggered on the event bus.
 func (b *BigTableVCS) StartTracking(ctx context.Context, evt eventbus.EventBus) {
 	if evt == nil {
 		sklog.Warningf("Not starting tracking eventbus was nil")
 		return
 	}

 	// Keep track of commits.
 	var prevCommits []*vcsinfo.IndexCommit
 	go util.RepeatCtx(defaultWatchInterval, ctx, func(ctx context.Context) {
 		allBranches, err := b.gitStore.GetBranches(ctx)
 		if err != nil {
 			sklog.Errorf("Error retrieving branches: %s", err)
 			return
 		}

 		branchInfo, ok := allBranches[b.defaultBranch]
 		if !ok {
 			sklog.Errorf("Branch %s not found in gitstore", b.defaultBranch)
 			return
 		}

 		startIdx := util.MaxInt(0, branchInfo.Index+1-nCommits)
 		commits, err := b.gitStore.RangeN(ctx, startIdx, int(math.MaxInt32), b.defaultBranch)
 		if err != nil {
 			sklog.Errorf("Error getting last %d commits: %s", nCommits, err)
 			return
 		}

 		// If we received new commits then publish an event and save them for the next round.
 		if len(prevCommits) != len(commits) || commits[len(commits)-1].Index > prevCommits[len(prevCommits)-1].Index {
 			prevCommits = commits
 			cpCommits := append([]*vcsinfo.IndexCommit{}, commits...)
 			evt.Publish(EV_NEW_GIT_COMMIT, cpCommits, false)
 		}
 	})
 }

 // Make sure BigTableVCS fulfills the VCS interface
 var _ vcsinfo.VCS = (*BigTableVCS)(nil)
	package bt_vcs

	import (
	"bytes"
	"context"
	"errors"
	"math"
	"sort"
	"sync"
	"time"

	"go.skia.org/infra/go/depot_tools"
	"go.skia.org/infra/go/eventbus"
	"go.skia.org/infra/go/gitiles"
	"go.skia.org/infra/go/gitstore"
	"go.skia.org/infra/go/gitstore/bt_gitstore"
	"go.skia.org/infra/go/skerr"
	"go.skia.org/infra/go/sklog"
	"go.skia.org/infra/go/util"
	"go.skia.org/infra/go/vcsinfo"
	"golang.org/x/sync/errgroup"
	)

	const (
	// EV_NEW_GIT_COMMIT is the event that is fired when a previously unseen Git commit is available.
	// The event data for this commit is of type []*vcsinfo.IndexCommit containing all new commits
	// that have been added since the last commit was sent.
	EV_NEW_GIT_COMMIT = "gitstore:new-git-commit"

	// defaultWatchInterval is the interval at which we check for new commits being added to the repo.
	defaultWatchInterval = time.Second * 10

	// nCommits is a parameter for StartTracking to tune how many
	// commits gets sent over the event bus when new commits are
	// detected.
	nCommits = 20
	)

	// BigTableVCS implements the vcsinfo.VCS interface based on a BT-backed GitStore.
	type BigTableVCS struct {
	gitStore gitstore.GitStore
	repo *gitiles.Repo
	defaultBranch string
	secondaryVCS vcsinfo.VCS
	secondaryExtractor depot_tools.DEPSExtractor

	branchInfo *gitstore.BranchPointer

	// This mutex protects detailsCache and indexCommits
	mutex sync.RWMutex
	// detailsCache is for LongCommits so we don't have to query gitStore every time
	detailsCache map[string]*vcsinfo.LongCommit
	indexCommits []*vcsinfo.IndexCommit
	}

	// NewVCS returns an instance of vcsinfo.VCS that is backed by the given GitStore and uses the
	// gittiles.Repo to retrieve files. Each instance provides an interface to one branch.
	// If defaultBranch is gitstore.ALL_BRANCHES all commits in the repository are considered.
	// The instances of gitiles.Repo is only used to fetch files.
	func New(ctx context.Context, gitStore gitstore.GitStore, defaultBranch string, repo gitiles.Repo) (BigTableVCS, error) {
	if gitStore == nil {
	return nil, errors.New("Cannot have nil gitStore")
	}
	ret := &BigTableVCS{
	gitStore: gitStore,
	repo: repo,
	defaultBranch: defaultBranch,
	detailsCache: map[string]*vcsinfo.LongCommit{},
	}
	if err := ret.Update(ctx, true, false); err != nil {
	return nil, skerr.Wrapf(err, "could not perform initial update")
	}

	return ret, nil
	}

	// GetBranch implements the vcsinfo.VCS interface.
	func (b *BigTableVCS) GetBranch() string {
	return b.defaultBranch
	}

	// SetSecondaryRepo allows to add a secondary repository and extractor to this instance.
	// It is not included in the constructor since it is currently only used by the Gold ingesters.
	func (b *BigTableVCS) SetSecondaryRepo(secVCS vcsinfo.VCS, extractor depot_tools.DEPSExtractor) {
	b.secondaryVCS = secVCS
	b.secondaryExtractor = extractor
	}

	// Update implements the vcsinfo.VCS interface
	func (b *BigTableVCS) Update(ctx context.Context, pull, allBranches bool) error {
	// Check if we need to pull across all branches.
	targetBranch := b.defaultBranch
	if allBranches {
	targetBranch = gitstore.ALL_BRANCHES
	}

	// Simulate a pull by fetching the latest head of the target branch.
	if pull {
	branchHeads, err := b.gitStore.GetBranches(ctx)
	if err != nil {
	return skerr.Wrap(err)
	}

	var ok bool
	b.branchInfo, ok = branchHeads[targetBranch]
	if !ok {
	r := (b.gitStore.(*bt_gitstore.BigTableGitStore)).RepoURL
	return skerr.Fmt("unable to find branch %q in BigTable repo %s", targetBranch, r)
	}
	}

	// Get all index commits for the current branch.
	if err := b.fetchIndexRange(ctx, 0, b.branchInfo.Index+1); err != nil {
	return skerr.Wrapf(err, "could not fetch index commits [0,%d]", b.branchInfo.Index+1)
	}
	// warm the cache of long commits
	hashes := func() []string {
	b.mutex.RLock()
	defer b.mutex.RUnlock()
	hashes := make([]string, 0, len(b.indexCommits))
	for _, ic := range b.indexCommits {
	hashes = append(hashes, ic.Hash)
	}
	return hashes
	}()

	_, err := b.DetailsMulti(ctx, hashes, false)
	return err
	}

	// From implements the vcsinfo.VCS interface
	func (b *BigTableVCS) From(start time.Time) []string {
	b.mutex.RLock()
	defer b.mutex.RUnlock()

	// Add a millisecond because we only want commits after the startTime. Timestamps in git are
	// only at second level granularity.
	found := b.timeRange(start.Add(time.Millisecond), vcsinfo.MaxTime)
	ret := make([]string, len(found))
	for i, c := range found {
	ret[i] = c.Hash
	}
	return ret
	}

	// Details implements the vcsinfo.VCS interface
	func (b BigTableVCS) Details(ctx context.Context, hash string, includeBranchInfo bool) (vcsinfo.LongCommit, error) {
	// Check the cache first, being sure to unlock the mutex after so that
	// if we have to poll BT for it, we aren't blocking the whole time.
	rv := func() *vcsinfo.LongCommit {
	b.mutex.RLock()
	defer b.mutex.RUnlock()
	c, ok := b.detailsCache[hash]
	if ok {
	// If the user is querying for branch info,
	// we have to check to see if the cached value has it.
	// If the cached value has it and the user didn't request it,
	// return it anyway.
	if !includeBranchInfo \|\| len(c.Branches) > 0 {
	return c
	}
	}
	return nil
	}()
	if rv != nil {
	return rv, nil
	}

	// cache miss, so query for it
	c, err := b.details(ctx, hash, includeBranchInfo)
	if err != nil {
	return nil, skerr.Wrapf(err, "could not fetch Details for commit %s", hash)
	}
	if c != nil {
	b.mutex.Lock()
	defer b.mutex.Unlock()
	b.detailsCache[hash] = c
	}
	return c, err
	}

	// DetailsMulti implements the vcsinfo.VCS interface
	func (b BigTableVCS) DetailsMulti(ctx context.Context, hashes []string, includeBranchInfo bool) ([]vcsinfo.LongCommit, error) {
	// Instantiate a list of N nil values so we can directly insert them into the proper index.
	rv := make([]*vcsinfo.LongCommit, len(hashes))

	missedHashes := []string{}
	// Index into hashes of which hashes were not in the cache
	missedHashesIdx := []int{}

	func() {
	b.mutex.RLock()
	defer b.mutex.RUnlock()
	for i, hash := range hashes {
	c, ok := b.detailsCache[hash]
	// If the user is querying for branch info,
	// we have to check to see if the cached value has it.
	if ok && (!includeBranchInfo \|\| len(c.Branches) > 0) {
	rv[i] = c
	} else {
	missedHashesIdx = append(missedHashesIdx, i)
	missedHashes = append(missedHashes, hash)
	}
	}
	}()
	if len(missedHashes) == 0 {
	return rv, nil
	}

	// bulk fetch the missedHashes. A simpler, but potentially slower, approach
	// could just do a b.details() for each commit (parallelized with an errgroup)
	commits, err := b.gitStore.Get(ctx, missedHashes)
	if err != nil {
	return nil, skerr.Wrapf(err, "Get missed hashes %q (superset %q)", missedHashes, hashes)
	}

	if includeBranchInfo {
	branchPointers, err := b.gitStore.GetBranches(ctx)
	if err != nil {
	return nil, skerr.Wrap(err)
	}

	// We need to do an individual request per commit for its branch info
	var egroup errgroup.Group
	for _, c := range commits {
	if c != nil {
	// Create a closure since we pass each value of 'c' to its own go-routine.
	func(c *vcsinfo.LongCommit) {
	egroup.Go(func() error {
	branches, err := b.getBranchInfo(ctx, c, branchPointers)
	if err != nil {
	return skerr.Wrapf(err, "getBranchInfo for commit %s", c.Hash)
	}
	c.Branches = branches
	return nil
	})
	}(c)
	}
	}
	if err := egroup.Wait(); err != nil {
	return nil, skerr.Wrapf(err, "Fetching branch info for %q", missedHashes)
	}
	}

	b.mutex.Lock()
	defer b.mutex.Unlock()
	for i, hash := range missedHashes {
	c := commits[i]
	if c != nil {
	rv[missedHashesIdx[i]] = c
	b.detailsCache[hash] = c
	}
	}

	return rv, nil
	}

	// details returns all meta data details we care about.
	func (b BigTableVCS) details(ctx context.Context, hash string, includeBranchInfo bool) (vcsinfo.LongCommit, error) {
	commits, err := b.gitStore.Get(ctx, []string{hash})
	if err != nil {
	return nil, err
	}

	if len(commits) == 0 \|\| commits[0] == nil {
	return nil, skerr.Fmt("Commit %s not found", hash)
	}

	if includeBranchInfo {
	branchPointers, err := b.gitStore.GetBranches(ctx)
	if err != nil {
	return nil, skerr.Wrap(err)
	}

	branches, err := b.getBranchInfo(ctx, commits[0], branchPointers)
	if err != nil {
	return nil, skerr.Wrapf(err, "getting branch info for commit %s", commits[0].Hash)
	}
	commits[0].Branches = branches
	}
	return commits[0], nil
	}

	// getBranchInfo determines which branches contain the given commit 'c'.
	// This function can potentially spawn a huge number of goroutines (one per branch).
	func (b BigTableVCS) getBranchInfo(ctx context.Context, c vcsinfo.LongCommit, allBranches map[string]*gitstore.BranchPointer) (map[string]bool, error) {
	ret := make(map[string]bool, len(allBranches))
	var mutex sync.Mutex
	var egroup errgroup.Group
	for branchName := range allBranches {
	if branchName != gitstore.ALL_BRANCHES {
	func(branchName string) {
	egroup.Go(func() error {
	// Since we cannot look up a commit in a branch directly we query for all commits that
	// occurred at that specific timestamp (Git has second granularity) on the target branch.
	// Then we check whether the target commit is returned as part of the result.
	commits, err := b.gitStore.RangeByTime(ctx, c.Timestamp, c.Timestamp.Add(time.Second), branchName)
	if err != nil {
	return skerr.Wrapf(err, "range query for branch %s", branchName)
	}

	// Iterate over the commits at the given timestamp. Most of the time there should
	// only be one commit at a given one second time range.
	for _, idxCommit := range commits {
	if idxCommit.Hash == c.Hash {
	mutex.Lock()
	ret[branchName] = true
	mutex.Unlock()
	break
	}
	}
	return nil
	})
	}(branchName)
	}
	}
	if err := egroup.Wait(); err != nil {
	return nil, skerr.Wrapf(err, "retrieving branch info for %s", c.Hash)
	}
	return ret, nil
	}

	// LastNIndex implements the vcsinfo.VCS interface
	func (b BigTableVCS) LastNIndex(N int) []vcsinfo.IndexCommit {
	b.mutex.RLock()
	defer b.mutex.RUnlock()

	if N > len(b.indexCommits) {
	N = len(b.indexCommits)
	}
	ret := make([]*vcsinfo.IndexCommit, 0, N)
	return append(ret, b.indexCommits[len(b.indexCommits)-N:]...)
	}

	// Range implements the vcsinfo.VCS interface
	func (b BigTableVCS) Range(begin, end time.Time) []vcsinfo.IndexCommit {
	return b.timeRange(begin, end)
	}

	// IndexOf implements the vcsinfo.VCS interface
	func (b *BigTableVCS) IndexOf(ctx context.Context, hash string) (int, error) {
	b.mutex.RLock()
	defer b.mutex.RUnlock()

	for i := len(b.indexCommits) - 1; i >= 0; i-- {
	if hash == b.indexCommits[i].Hash {
	return b.indexCommits[i].Index, nil
	}
	}

	return -1, skerr.Fmt("commit %s not found", hash)
	}

	// ByIndex implements the vcsinfo.VCS interface
	func (b BigTableVCS) ByIndex(ctx context.Context, N int) (vcsinfo.LongCommit, error) {
	// findFn returns the hash when N is within commits
	findFn := func(commits []*vcsinfo.IndexCommit) string {
	i := sort.Search(len(commits), func(i int) bool { return commits[i].Index >= N })
	return commits[i].Hash
	}

	hash := func() string {
	b.mutex.RLock()
	defer b.mutex.RUnlock()
	if len(b.indexCommits) > 0 {
	firstIdx := b.indexCommits[0].Index
	lastIdx := b.indexCommits[len(b.indexCommits)-1].Index
	if (N >= firstIdx) && (N <= lastIdx) {
	return findFn(b.indexCommits)
	}
	}
	return ""
	}()

	// Fetch the hash
	if hash == "" {
	return nil, skerr.Fmt("Hash index not found: %d", N)
	}
	return b.Details(ctx, hash, false)
	}

	// GetFile implements the vcsinfo.VCS interface
	func (b *BigTableVCS) GetFile(ctx context.Context, fileName, commitHash string) (string, error) {
	var buf bytes.Buffer
	if err := b.repo.ReadFileAtRef(ctx, fileName, commitHash, &buf); err != nil {
	return "", skerr.Wrapf(err, "reading file %s @ %s via gitiles", fileName, commitHash)
	}
	return buf.String(), nil
	}

	// ResolveCommit implements the vcsinfo.VCS interface
	func (b *BigTableVCS) ResolveCommit(ctx context.Context, commitHash string) (string, error) {
	if b.secondaryVCS == nil {
	return "", vcsinfo.NoSecondaryRepo
	}

	foundCommit, err := b.secondaryExtractor.ExtractCommit(b.secondaryVCS.GetFile(ctx, "DEPS", commitHash))
	if err != nil {
	return "", err
	}
	return foundCommit, nil
	}

	// GetGitStore implements the gitstore.GitStoreBased interface
	func (b *BigTableVCS) GetGitStore() gitstore.GitStore {
	return b.gitStore
	}

	// fetchIndexRange gets in the range [startIndex, endIndex).
	func (b *BigTableVCS) fetchIndexRange(ctx context.Context, startIndex, endIndex int) error {
	newIC, err := b.gitStore.RangeN(ctx, startIndex, endIndex, b.defaultBranch)
	if err != nil {
	return err
	}

	if len(newIC) == 0 {
	return nil
	}

	b.mutex.Lock()
	defer b.mutex.Unlock()
	b.indexCommits = newIC
	return nil
	}

	func (b BigTableVCS) timeRange(start time.Time, end time.Time) []vcsinfo.IndexCommit {
	n := len(b.indexCommits)
	startIdx := 0
	for ; startIdx < n; startIdx++ {
	exp := b.indexCommits[startIdx].Timestamp.After(start) \|\| b.indexCommits[startIdx].Timestamp.Equal(start)
	if exp {
	break
	}
	}

	endIdx := startIdx
	for ; endIdx < n; endIdx++ {
	exp := b.indexCommits[endIdx].Timestamp.After(end) \|\| b.indexCommits[endIdx].Timestamp.Equal(end)
	if exp {
	break
	}
	}

	if endIdx <= startIdx {
	return []*vcsinfo.IndexCommit{}
	}
	return b.indexCommits[startIdx:endIdx]
	}

	// StartTracking begins watching the repo for changes on a background thread.
	// When a new commit is detected a EV_NEW_GIT_COMMIT event is triggered on the event bus.
	func (b *BigTableVCS) StartTracking(ctx context.Context, evt eventbus.EventBus) {
	if evt == nil {
	sklog.Warningf("Not starting tracking eventbus was nil")
	return
	}

	// Keep track of commits.
	var prevCommits []*vcsinfo.IndexCommit
	go util.RepeatCtx(defaultWatchInterval, ctx, func(ctx context.Context) {
	allBranches, err := b.gitStore.GetBranches(ctx)
	if err != nil {
	sklog.Errorf("Error retrieving branches: %s", err)
	return
	}

	branchInfo, ok := allBranches[b.defaultBranch]
	if !ok {
	sklog.Errorf("Branch %s not found in gitstore", b.defaultBranch)
	return
	}

	startIdx := util.MaxInt(0, branchInfo.Index+1-nCommits)
	commits, err := b.gitStore.RangeN(ctx, startIdx, int(math.MaxInt32), b.defaultBranch)
	if err != nil {
	sklog.Errorf("Error getting last %d commits: %s", nCommits, err)
	return
	}

	// If we received new commits then publish an event and save them for the next round.
	if len(prevCommits) != len(commits) \|\| commits[len(commits)-1].Index > prevCommits[len(prevCommits)-1].Index {
	prevCommits = commits
	cpCommits := append([]*vcsinfo.IndexCommit{}, commits...)
	evt.Publish(EV_NEW_GIT_COMMIT, cpCommits, false)
	}
	})
	}

	// Make sure BigTableVCS fulfills the VCS interface
	var _ vcsinfo.VCS = (*BigTableVCS)(nil)