syncer.go

package repos

import (
	"context"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/sourcegraph/log"
	"go.opentelemetry.io/otel/attribute"
	"golang.org/x/sync/singleflight"
	"golang.org/x/time/rate"

	"github.com/sourcegraph/sourcegraph/internal/actor"
	"github.com/sourcegraph/sourcegraph/internal/api"
	"github.com/sourcegraph/sourcegraph/internal/conf"
	"github.com/sourcegraph/sourcegraph/internal/database"
	"github.com/sourcegraph/sourcegraph/internal/dotcom"
	"github.com/sourcegraph/sourcegraph/internal/errcode"
	"github.com/sourcegraph/sourcegraph/internal/goroutine"
	"github.com/sourcegraph/sourcegraph/internal/licensing"
	"github.com/sourcegraph/sourcegraph/internal/metrics"
	"github.com/sourcegraph/sourcegraph/internal/observation"
	"github.com/sourcegraph/sourcegraph/internal/trace"
	"github.com/sourcegraph/sourcegraph/internal/types"
	"github.com/sourcegraph/sourcegraph/lib/errors"
)

// A Syncer periodically synchronizes available repositories from all its given Sources
// with the stored Repositories in Sourcegraph.
type Syncer struct {
	Sourcer Sourcer
	Store   Store

	// Synced is sent a collection of Repos that were synced by Sync (only if Synced is non-nil)
	Synced chan types.RepoSyncDiff

	ObsvCtx *observation.Context

	// Now is time.Now. Can be set by tests to get deterministic output.
	Now func() time.Time

	// Ensure that we only run one sync per repo at a time
	syncGroup singleflight.Group
}

func NewSyncer(observationCtx *observation.Context, store Store, sourcer Sourcer) *Syncer {
	return &Syncer{
		Sourcer: sourcer,
		Store:   store,
		Synced:  make(chan types.RepoSyncDiff),
		Now:     func() time.Time { return time.Now().UTC() },
		ObsvCtx: observation.ContextWithLogger(observationCtx.Logger.Scoped("syncer"), observationCtx),
	}
}

// RunOptions contains options customizing Run behaviour.
type RunOptions struct {
	EnqueueInterval func() time.Duration // Defaults to 1 minute
	MinSyncInterval func() time.Duration // Defaults to 1 minute
	DequeueInterval time.Duration        // Default to 10 seconds
}

// Routines returns the goroutines that run the Sync at the specified interval.
func (s *Syncer) Routines(ctx context.Context, store Store, opts RunOptions) []goroutine.BackgroundRoutine {
	if opts.EnqueueInterval == nil {
		opts.EnqueueInterval = func() time.Duration { return time.Minute }
	}
	if opts.MinSyncInterval == nil {
		opts.MinSyncInterval = func() time.Duration { return time.Minute }
	}
	if opts.DequeueInterval == 0 {
		opts.DequeueInterval = 10 * time.Second
	}

	worker, resetter, syncerJanitor := NewSyncWorker(ctx, observation.ContextWithLogger(s.ObsvCtx.Logger.Scoped("syncWorker"), s.ObsvCtx),
		store.Handle(),
		&syncHandler{
			syncer:          s,
			store:           store,
			minSyncInterval: opts.MinSyncInterval,
		}, SyncWorkerOptions{
			WorkerInterval: opts.DequeueInterval,
			NumHandlers:    conf.RepoConcurrentExternalServiceSyncers(),
			CleanupOldJobs: true,
		},
	)

	scheduler := goroutine.NewPeriodicGoroutine(
		actor.WithInternalActor(ctx),
		goroutine.HandlerFunc(func(ctx context.Context) error {
			if conf.Get().DisableAutoCodeHostSyncs {
				return nil
			}

			if err := store.EnqueueSyncJobs(ctx); err != nil {
				return errors.Wrap(err, "enqueueing sync jobs")
			}

			return nil
		}),
		goroutine.WithName("repo-updater.repo-sync-scheduler"),
		goroutine.WithDescription("enqueues sync jobs for external service sync jobs"),
		goroutine.WithIntervalFunc(opts.EnqueueInterval),
	)

	routines := []goroutine.BackgroundRoutine{worker, resetter, syncerJanitor, scheduler}

	return routines
}

type syncHandler struct {
	syncer          *Syncer
	store           Store
	minSyncInterval func() time.Duration
}

func (s *syncHandler) Handle(ctx context.Context, _ log.Logger, sj *SyncJob) (err error) {
	// Limit calls to progressRecorder as it will most likely hit the database
	progressLimiter := rate.NewLimiter(rate.Limit(1.0), 1)

	progressRecorder := func(ctx context.Context, progress SyncProgress, final bool) error {
		if final || progressLimiter.Allow() {
			return s.store.ExternalServiceStore().UpdateSyncJobCounters(ctx, &types.ExternalServiceSyncJob{
				ID:              int64(sj.ID),
				ReposSynced:     progress.Synced,
				RepoSyncErrors:  progress.Errors,
				ReposAdded:      progress.Added,
				ReposDeleted:    progress.Deleted,
				ReposModified:   progress.Modified,
				ReposUnmodified: progress.Unmodified,
			})
		}
		return nil
	}

	return s.syncer.SyncExternalService(ctx, sj.ExternalServiceID, s.minSyncInterval(), progressRecorder)
}

// TriggerExternalServiceSync will enqueue a sync job for the supplied external
// service
func (s *Syncer) TriggerExternalServiceSync(ctx context.Context, id int64) error {
	return s.Store.EnqueueSingleSyncJob(ctx, id)
}

const (
	ownerUndefined = ""
	ownerSite      = "site"
)

type ErrUnauthorized struct{}

func (e ErrUnauthorized) Error() string {
	return "bad credentials"
}

func (e ErrUnauthorized) Unauthorized() bool {
	return true
}

type ErrForbidden struct{}

func (e ErrForbidden) Error() string {
	return "forbidden"
}

func (e ErrForbidden) Forbidden() bool {
	return true
}

type ErrAccountSuspended struct{}

func (e ErrAccountSuspended) Error() string {
	return "account suspended"
}

func (e ErrAccountSuspended) AccountSuspended() bool {
	return true
}

func (s *Syncer) notifyDeleted(ctx context.Context, deleted ...api.RepoID) {
	var d types.RepoSyncDiff
	for _, id := range deleted {
		d.Deleted = append(d.Deleted, &types.Repo{ID: id})
	}
	observeDiff(d)

	if s.Synced != nil && d.Len() > 0 {
		select {
		case <-ctx.Done():
		case s.Synced <- d:
		}
	}
}

// SyncProgress represents running counts for an external service sync.
type SyncProgress struct {
	Synced int32 `json:"synced,omitempty"`
	Errors int32 `json:"errors,omitempty"`

	// Diff stats
	Added      int32 `json:"added,omitempty"`
	Removed    int32 `json:"removed,omitempty"`
	Modified   int32 `json:"modified,omitempty"`
	Unmodified int32 `json:"unmodified,omitempty"`

	Deleted int32 `json:"deleted,omitempty"`
}

type LicenseError struct {
	error
}

// progressRecorderFunc is a function that implements persisting sync progress.
// The final param represents whether this is the final call. This allows the
// function to decide whether to drop some intermediate calls.
type progressRecorderFunc func(ctx context.Context, progress SyncProgress, final bool) error

// SyncExternalService syncs repos using the supplied external service in a streaming fashion, rather than batch.
// This allows very large sync jobs (i.e. that source potentially millions of repos) to incrementally persist changes.
// Deletes of repositories that were not sourced are done at the end.
func (s *Syncer) SyncExternalService(
	ctx context.Context,
	externalServiceID int64,
	minSyncInterval time.Duration,
	progressRecorder progressRecorderFunc,
) (err error) {
	logger := s.ObsvCtx.Logger.Scoped("SyncExternalService").With(log.Int64("externalServiceID", externalServiceID))
	logger.Info("syncing external service")

	// Ensure the job field is recorded when monitoring external API calls
	ctx = metrics.ContextWithTask(ctx, "SyncExternalService")

	var svc *types.ExternalService
	ctx, save := s.observeSync(ctx, "Syncer.SyncExternalService")
	defer func() { save(svc, err) }()

	// We don't use tx here as the sourcing process below can be slow and we don't
	// want to hold a lock on the external_services table for too long.
	svc, err = s.Store.ExternalServiceStore().GetByID(ctx, externalServiceID)
	if err != nil {
		return errors.Wrap(err, "fetching external services")
	}

	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	// From this point we always want to make a best effort attempt to update the
	// service timestamps
	var modified bool
	defer func() {
		now := s.Now()
		interval := calcSyncInterval(now, svc.LastSyncAt, minSyncInterval, modified, err)

		nextSyncAt := now.Add(interval)
		lastSyncAt := now

		// We call Update here instead of Upsert, because upsert stores all fields of the external
		// service, and syncs take a while so changes to the external service made while this sync
		// was running would be overwritten again.
		if err := s.Store.ExternalServiceStore().Update(ctx, nil, svc.ID, &database.ExternalServiceUpdate{
			LastSyncAt: &lastSyncAt,
			NextSyncAt: &nextSyncAt,
		}); err != nil {
			// We only want to log this error, not return it
			logger.Error("upserting external service", log.Error(err))
		}

		logger.Debug("synced external service", log.Duration("backoff duration", interval))
	}()

	src, err := s.Sourcer(ctx, svc)
	if err != nil {
		return err
	}

	if err := src.CheckConnection(ctx); err != nil {
		logger.Warn("connection check failed. syncing repositories might still succeed.", log.Error(err))
	}

	results := make(chan SourceResult)
	go func() {
		src.ListRepos(ctx, results)
		close(results)
	}()

	seen := make(map[api.RepoID]struct{})
	var errs error
	fatal := func(err error) bool {
		// If the error is just a warning, then it is not fatal.
		if errors.IsWarning(err) && !errcode.IsAccountSuspended(err) {
			return false
		}

		return errcode.IsUnauthorized(err) ||
			errcode.IsForbidden(err) ||
			errcode.IsAccountSuspended(err)
	}

	logger = s.ObsvCtx.Logger.With(log.Object("svc", log.String("name", svc.DisplayName), log.Int64("id", svc.ID)))

	var syncProgress SyncProgress
	// Record the final progress state
	defer func() {
		// Use a different context here so that we make sure to record progress
		// even if context has been canceled
		if err := progressRecorder(context.Background(), syncProgress, true); err != nil {
			logger.Error("recording final sync progress", log.Error(err))
		}
	}()

	// Insert or update repos as they are sourced. Keep track of what was seen so we
	// can remove anything else at the end.
	for res := range results {
		logger.Debug("received result", log.String("repo", fmt.Sprintf("%v", res)))

		if err := progressRecorder(ctx, syncProgress, false); err != nil {
			logger.Warn("recording sync progress", log.Error(err))
		}

		if err := res.Err; err != nil {
			syncProgress.Errors++
			logger.Error("error from codehost", log.Int("seen", len(seen)), log.Error(err))

			errs = errors.Append(errs, errors.Wrapf(err, "fetching from code host %s", svc.DisplayName))
			if fatal(err) {
				// Delete all external service repos of this external service
				logger.Error("stopping external service sync due to fatal error from codehost", log.Error(err))
				seen = map[api.RepoID]struct{}{}
				break
			}

			continue
		}

		sourced := res.Repo

		if dotcom.SourcegraphDotComMode() && sourced.Private {
			err := errors.Newf("%s is private, but dotcom does not support private repositories.", sourced.Name)
			syncProgress.Errors++
			logger.Error("failed to sync private repo", log.String("repo", string(sourced.Name)), log.Error(err))
			errs = errors.Append(errs, err)
			continue
		}

		var diff types.RepoSyncDiff
		if diff, err = s.sync(ctx, svc, sourced); err != nil {
			syncProgress.Errors++
			logger.Error("failed to sync, skipping", log.String("repo", string(sourced.Name)), log.Error(err))
			errs = errors.Append(errs, err)
			continue
		}

		syncProgress.Added += int32(diff.Added.Len())
		syncProgress.Removed += int32(diff.Deleted.Len())
		syncProgress.Modified += int32(diff.Modified.Repos().Len())
		syncProgress.Unmodified += int32(diff.Unmodified.Len())

		for _, r := range diff.Repos() {
			seen[r.ID] = struct{}{}
		}
		syncProgress.Synced = int32(len(seen))

		modified = modified || len(diff.Modified)+len(diff.Added) > 0
	}

	// This information isn't necessarily an error in and of itself, but it's possible
	// for the code host to misbehave in a way that returns a legitimate-looking response
	// (e.g. a non-fatal HTTP status code, with a well-formed response body) that is
	// nonetheless incorrect (e.g. temporarily returning empty set of repositories
	// instead of an error).
	//
	// Because of this, we have to be able to log every response from the code host
	// so that we can audit them later so that we can rule in/out the above scenario.
	// So, we choose the WARN level instead of INFO so that this info is logged by default.
	logger.Warn("finished listing repositories from external service",
		log.Object("syncProgress",
			log.Int32("synced", syncProgress.Synced),
			log.Int32("errors", syncProgress.Errors),
			log.Int32("added", syncProgress.Added),
			log.Int32("removed", syncProgress.Removed),
			log.Int32("modified", syncProgress.Modified),
			log.Int32("unmodified", syncProgress.Unmodified),
		),
		log.Int("seen", len(seen)),
		log.Bool("modified", modified),
		log.Error(errs),
	)

	// We don't delete any repos of site-level external services if there were any
	// non-warning errors during a sync.
	//
	// Only user or organization external services will delete
	// repos in a sync run with fatal errors.
	//
	// Site-level external services can own lots of repos and are managed by site admins.
	// It's preferable to have them fix any invalidated token manually rather than deleting the repos automatically.
	deleted := 0

	// If all of our errors are warnings and either Forbidden or Unauthorized,
	// we want to proceed with the deletion. This is to be able to properly sync
	// repos (by removing ones if code-host permissions have changed).
	abortDeletion := false
	if errs != nil {
		logger.Error("received errors during sync", log.Error(errs))
		var ref errors.MultiError
		if errors.As(errs, &ref) {
			for _, e := range ref.Errors() {
				if errors.IsWarning(e) {
					baseError := errors.Unwrap(e)
					if !errcode.IsForbidden(baseError) && !errcode.IsUnauthorized(baseError) {
						abortDeletion = true
						logger.Info("aborting deletion due to fatal error", log.Error(e))
						break
					}
					continue
				}
				if errors.As(e, &LicenseError{}) {
					continue
				}
				abortDeletion = true
				logger.Info("aborting deletion due to fatal error", log.Error(e))
				break
			}
		}
	}

	if !abortDeletion {
		// Remove associations and any repos that are no longer associated with any
		// external service.
		//
		// We don't want to delete all repos that weren't seen if we had a lot of
		// spurious errors since that could cause lots of repos to be deleted, only to be
		// added the next sync. We delete only if we had no errors,
		// or all of our errors are warnings and either Forbidden or Unauthorized,
		// or we had one of the fatal errors and the service is not site owned.
		var deletedErr error
		deleted, deletedErr = s.delete(ctx, svc, seen)
		if deletedErr != nil {
			logger.Warn("failed to delete some repos",
				log.Int("seen", len(seen)),
				log.Int("deleted", deleted),
				log.Error(deletedErr),
			)

			errs = errors.Append(errs, errors.Wrap(deletedErr, "some repos couldn't be deleted"))
		}

		if deleted > 0 {
			syncProgress.Deleted += int32(deleted)
			logger.Warn("deleted not seen repos",
				log.Int("seen", len(seen)),
				log.Int("deleted", deleted),
				log.Error(err),
			)
		}
	}

	modified = modified || deleted > 0

	return errs
}

// syncs a sourced repo of a given external service, returning a diff with a single repo.
func (s *Syncer) sync(ctx context.Context, svc *types.ExternalService, sourced *types.Repo) (d types.RepoSyncDiff, err error) {
	tx, err := s.Store.Transact(ctx)
	if err != nil {
		return types.RepoSyncDiff{}, errors.Wrap(err, "syncer: opening transaction")
	}

	defer func() {
		observeDiff(d)
		// We must commit the transaction before publishing to s.Synced
		// so that gitserver finds the repo in the database.

		s.ObsvCtx.Logger.Debug("committing transaction")
		err = tx.Done(err)
		if err != nil {
			s.ObsvCtx.Logger.Warn("failed to commit transaction", log.Error(err))
			return
		}

		if s.Synced != nil && d.Len() > 0 {
			select {
			case <-ctx.Done():
				s.ObsvCtx.Logger.Debug("sync context done")
			case s.Synced <- d:
				s.ObsvCtx.Logger.Debug("diff synced")
			}
		}
	}()

	stored, err := tx.RepoStore().List(ctx, database.ReposListOptions{
		Names:          []string{string(sourced.Name)},
		ExternalRepos:  []api.ExternalRepoSpec{sourced.ExternalRepo},
		IncludeBlocked: true,
		IncludeDeleted: true,
		UseOr:          true,
	})
	if err != nil {
		return types.RepoSyncDiff{}, errors.Wrap(err, "syncer: getting repo from the database")
	}

	switch len(stored) {
	case 2: // Existing repo with a naming conflict
		// Scenario where this can happen:
		// 1. Repo `owner/repo1` with external_id 1 exists
		// 2. Repo `owner/repo2` with external_id 2 exists
		// 3. The owner deletes repo1, and renames repo2 to repo1
		// 4. We sync and we receive `owner/repo1` with external_id 2
		//
		// Then the above query will return two results: one matching the name owner/repo1, and one matching the external_service_id 2
		// The original owner/repo1 should be deleted, and then owner/repo2 with the matching external_service_id should be updated
		s.ObsvCtx.Logger.Debug("naming conflict")

		// Pick this sourced repo to own the name by deleting the other repo. If it still exists, it'll have a different
		// name when we source it from the same code host, and it will be re-created.
		var conflicting, existing *types.Repo
		for _, r := range stored {
			if r.ExternalRepo.Equal(&sourced.ExternalRepo) {
				existing = r
			} else {
				conflicting = r
			}
		}

		// invariant: conflicting can't be nil due to our database constraints
		if err = tx.RepoStore().Delete(ctx, conflicting.ID); err != nil {
			return types.RepoSyncDiff{}, errors.Wrap(err, "syncer: failed to delete conflicting repo")
		}

		// We fallthrough to the next case after removing the conflicting repo in order to update
		// the winner (i.e. existing). This works because we mutate stored to contain it, which the case expects.
		stored = types.Repos{existing}
		s.ObsvCtx.Logger.Debug("retrieved stored repo, falling through", log.String("stored", fmt.Sprintf("%v", stored)))
		fallthrough
	case 1: // Existing repo, update.
		wasDeleted := !stored[0].DeletedAt.IsZero()
		s.ObsvCtx.Logger.Debug("existing repo")
		if err := UpdateRepoLicenseHook(ctx, tx, stored[0], sourced); err != nil {
			return types.RepoSyncDiff{}, LicenseError{errors.Wrapf(err, "syncer: failed to update repo %s", sourced.Name)}
		}
		modified := stored[0].Update(sourced)
		if modified == types.RepoUnmodified {
			d.Unmodified = append(d.Unmodified, stored[0])
			break
		}

		if err = tx.UpdateExternalServiceRepo(ctx, svc, stored[0]); err != nil {
			return types.RepoSyncDiff{}, errors.Wrap(err, "syncer: failed to update external service repo")
		}

		*sourced = *stored[0]
		if wasDeleted {
			d.Added = append(d.Added, stored[0])
			s.ObsvCtx.Logger.Debug("revived soft-deleted repo")
		} else {
			d.Modified = append(d.Modified, types.RepoModified{Repo: stored[0], Modified: modified})
			s.ObsvCtx.Logger.Debug("appended to modified repos")
		}
	case 0: // New repo, create.
		s.ObsvCtx.Logger.Debug("new repo")

		if err := CreateRepoLicenseHook(ctx, tx, sourced); err != nil {
			return types.RepoSyncDiff{}, LicenseError{errors.Wrapf(err, "syncer: failed to update repo %s", sourced.Name)}
		}

		if err = tx.CreateExternalServiceRepo(ctx, svc, sourced); err != nil {
			return types.RepoSyncDiff{}, errors.Wrapf(err, "syncer: failed to create external service repo: %s", sourced.Name)
		}

		d.Added = append(d.Added, sourced)
		s.ObsvCtx.Logger.Debug("appended to added repos")
	default: // Impossible since we have two separate unique constraints on name and external repo spec
		panic("unreachable")
	}

	s.ObsvCtx.Logger.Debug("completed")
	return d, nil
}

// CreateRepoLicenseHook checks if there is still room for private repositories
// available in the applied license before creating a new private repository.
func CreateRepoLicenseHook(ctx context.Context, s Store, repo *types.Repo) error {
	// If the repository is public, we don't have to check anything
	if !repo.Private {
		return nil
	}

	if prFeature := (&licensing.FeaturePrivateRepositories{}); licensing.Check(prFeature) == nil {
		if prFeature.Unrestricted {
			return nil
		}

		numPrivateRepos, err := s.RepoStore().Count(ctx, database.ReposListOptions{OnlyPrivate: true})
		if err != nil {
			return err
		}

		if numPrivateRepos >= prFeature.MaxNumPrivateRepos {
			return errors.Newf("maximum number of private repositories included in license (%d) reached", prFeature.MaxNumPrivateRepos)
		}

		return nil
	}

	return licensing.NewFeatureNotActivatedError("The private repositories feature is not activated for this license. Please upgrade your license to use this feature.")
}

// UpdateRepoLicenseHook checks if there is still room for private repositories
// available in the applied license before updating a repository from public to private,
// or undeleting a private repository.
func UpdateRepoLicenseHook(ctx context.Context, s Store, existingRepo *types.Repo, newRepo *types.Repo) error {
	// If it is being updated to a public repository, or if a repository is being deleted, we don't have to check anything
	if !newRepo.Private || !newRepo.DeletedAt.IsZero() {
		return nil
	}

	if prFeature := (&licensing.FeaturePrivateRepositories{}); licensing.Check(prFeature) == nil {
		if prFeature.Unrestricted {
			return nil
		}

		numPrivateRepos, err := s.RepoStore().Count(ctx, database.ReposListOptions{OnlyPrivate: true})
		if err != nil {
			return err
		}

		if numPrivateRepos > prFeature.MaxNumPrivateRepos {
			return errors.Newf("maximum number of private repositories included in license (%d) reached", prFeature.MaxNumPrivateRepos)
		} else if numPrivateRepos == prFeature.MaxNumPrivateRepos {
			// If the repository is already private, we don't have to check anything
			newPrivateRepo := (!existingRepo.DeletedAt.IsZero() || !existingRepo.Private) && newRepo.Private // If restoring a deleted repository, or if it was a public repository, and is now private
			if newPrivateRepo {
				return errors.Newf("maximum number of private repositories included in license (%d) reached", prFeature.MaxNumPrivateRepos)
			}
		}

		return nil
	}

	return licensing.NewFeatureNotActivatedError("The private repositories feature is not activated for this license. Please upgrade your license to use this feature.")
}

func (s *Syncer) delete(ctx context.Context, svc *types.ExternalService, seen map[api.RepoID]struct{}) (int, error) {
	// We do deletion in a best effort manner, returning any errors for individual repos that failed to be deleted.
	deleted, err := s.Store.DeleteExternalServiceReposNotIn(ctx, svc, seen)

	s.notifyDeleted(ctx, deleted...)

	return len(deleted), err
}

func observeDiff(diff types.RepoSyncDiff) {
	for state, repos := range map[string]types.Repos{
		"added":      diff.Added,
		"modified":   diff.Modified.Repos(),
		"deleted":    diff.Deleted,
		"unmodified": diff.Unmodified,
	} {
		syncedTotal.WithLabelValues(state).Add(float64(len(repos)))
	}
}

func calcSyncInterval(
	now time.Time,
	lastSync time.Time,
	minSyncInterval time.Duration,
	modified bool,
	err error,
) time.Duration {
	const maxSyncInterval = 8 * time.Hour

	// Special case, we've never synced
	if err == nil && (lastSync.IsZero() || modified) {
		return minSyncInterval
	}

	// No change or there were errors, back off
	interval := now.Sub(lastSync) * 2
	if interval < minSyncInterval {
		return minSyncInterval
	}

	if interval > maxSyncInterval {
		return maxSyncInterval
	}

	return interval
}

func (s *Syncer) observeSync(
	ctx context.Context,
	name string,
	attributes ...attribute.KeyValue,
) (context.Context, func(*types.ExternalService, error)) {
	began := s.Now()
	tr, ctx := trace.New(ctx, name, attributes...)

	return ctx, func(svc *types.ExternalService, err error) {
		var owner string
		if svc == nil {
			owner = ownerUndefined
		} else {
			owner = ownerSite
		}

		syncStarted.WithLabelValues(name, owner).Inc()

		now := s.Now()
		took := now.Sub(began).Seconds()

		lastSync.WithLabelValues(name).Set(float64(now.Unix()))

		success := err == nil
		syncDuration.WithLabelValues(strconv.FormatBool(success), name).Observe(took)

		if !success {
			tr.SetError(err)
			syncErrors.WithLabelValues(name, owner, syncErrorReason(err)).Inc()
		}

		tr.End()
	}
}

func syncErrorReason(err error) string {
	switch {
	case err == nil:
		return ""
	case errcode.IsNotFound(err):
		return "not_found"
	case errcode.IsUnauthorized(err):
		return "unauthorized"
	case errcode.IsForbidden(err):
		return "forbidden"
	case errcode.IsTemporary(err):
		return "temporary"
	case strings.Contains(err.Error(), "expected path in npm/(scope/)?name"):
		// This is a known issue which we can filter out for now
		return "invalid_npm_path"
	case strings.Contains(err.Error(), "internal rate limit exceeded"):
		// We want to identify these as it's not an issue communicating with the code
		// host and is most likely caused by temporary traffic spikes.
		return "internal_rate_limit"
	default:
		return "unknown"
	}
}