func (p *processor) Process(ctx context.Context, store store.Store, upload store.Upload) (_ bool, err error) {
	// Ensure that the repo and revision are resolvable. If the repo does not exist, or if the repo has finished
	// cloning and the revision does not exist, then the upload will fail to process. If the repo is currently
	// cloning, then we'll requeue the upload to be tried again later. This will not increase the reset count
	// of the record (so this doesn't count against the upload as a legitimate attempt).
	if cloneInProgress, err := p.isRepoCurrentlyCloning(ctx, upload.RepositoryID, upload.Commit); err != nil {
		return false, err
	} else if cloneInProgress {
		if err := store.Requeue(ctx, upload.ID, time.Now().UTC().Add(CloneInProgressDelay)); err != nil {
			return false, errors.Wrap(err, "store.Requeue")
		}

		return true, nil
	}

	// Create scratch directory that we can clean on completion/failure
	tempDir, err := ioutil.TempDir("", "")
	if err != nil {
		return false, err
	}
	defer func() {
		if cleanupErr := os.RemoveAll(tempDir); cleanupErr != nil {
			log15.Warn("Failed to remove temporary directory", "path", tempDir, "err", cleanupErr)
		}
	}()

	// Pull raw uploaded data from bundle manager
	r, err := p.bundleManagerClient.GetUpload(ctx, upload.ID)
	if err != nil {
		return false, errors.Wrap(err, "bundleManager.GetUpload")
	}
	defer func() {
		if err != nil {
			// Remove upload file on error instead of waiting for it to expire
			if deleteErr := p.bundleManagerClient.DeleteUpload(ctx, upload.ID); deleteErr != nil {
				log15.Warn("Failed to delete upload file", "err", err)
			}
		}
	}()

	getChildren := func(ctx context.Context, dirnames []string) (map[string][]string, error) {
		directoryChildren, err := p.gitserverClient.DirectoryChildren(ctx, store, upload.RepositoryID, upload.Commit, dirnames)
		if err != nil {
			return nil, errors.Wrap(err, "gitserverClient.DirectoryChildren")
		}
		return directoryChildren, nil
	}

	groupedBundleData, err := correlation.Correlate(ctx, r, upload.ID, upload.Root, getChildren, p.metrics)
	if err != nil {
		return false, errors.Wrap(err, "correlation.Correlate")
	}

	if err := p.write(ctx, tempDir, groupedBundleData); err != nil {
		return false, err
	}

	// At this point we haven't touched the database. We're going to start a nested transaction
	// with Postgres savepoints. In the event that something after this point fails, we want to
	// update the upload record with an error message but do not want to alter any other data in
	// the database. Rolling back to this savepoint will allow us to discard any other changes
	// but still commit the transaction as a whole.
	savepointID, err := store.Savepoint(ctx)
	if err != nil {
		return false, errors.Wrap(err, "store.Savepoint")
	}
	defer func() {
		if err != nil {
			if rollbackErr := store.RollbackToSavepoint(ctx, savepointID); rollbackErr != nil {
				err = multierror.Append(err, rollbackErr)
			}
		}
	}()

	if err := p.updateXrepoData(ctx, store, upload, groupedBundleData.Packages, groupedBundleData.PackageReferences); err != nil {
		return false, err
	}

	// Send converted database file to bundle manager
	if err := p.sendDB(ctx, upload.ID, tempDir); err != nil {
		return false, err
	}

	return false, nil
}
// process converts a raw upload into a dump within the given transaction context. Returns true if the
// upload record was requeued and false otherwise.
process converts a raw upload into a dump within the given transaction context. Returns true if the
upload record was requeued and false otherwise.
// process converts a raw upload into a dump within the given transaction context . Returns true if the // upload record was requeued and false otherwise .
func (p *processor) Process(ctx context.Context, store store.Store, upload store.Upload) (_ bool, err error) {
	if cloneInProgress, err := p.isRepoCurrentlyCloning(ctx, upload.RepositoryID, upload.Commit); err != nil {
		return false, err
	} else if cloneInProgress {
		if err := store.Requeue(ctx, upload.ID, time.Now().UTC().Add(CloneInProgressDelay)); err != nil {
			return false, errors.Wrap(err, "store.Requeue")
		}
		return true, nil
	}
	tempDir, err := ioutil.TempDir("", "")
	if err != nil {
		return false, err
	}
	defer func() {
		if cleanupErr := os.RemoveAll(tempDir); cleanupErr != nil {
			log15.Warn("Failed to remove temporary directory", "path", tempDir, "err", cleanupErr)
		}
	}()
	r, err := p.bundleManagerClient.GetUpload(ctx, upload.ID)
	if err != nil {
		return false, errors.Wrap(err, "bundleManager.GetUpload")
	}
	defer func() {
		if err != nil {
			if deleteErr := p.bundleManagerClient.DeleteUpload(ctx, upload.ID); deleteErr != nil {
				log15.Warn("Failed to delete upload file", "err", err)
			}
		}
	}()
	getChildren := func(ctx context.Context, dirnames []string) (map[string][]string, error) {
		directoryChildren, err := p.gitserverClient.DirectoryChildren(ctx, store, upload.RepositoryID, upload.Commit, dirnames)
		if err != nil {
			return nil, errors.Wrap(err, "gitserverClient.DirectoryChildren")
		}
		return directoryChildren, nil
	}
	groupedBundleData, err := correlation.Correlate(ctx, r, upload.ID, upload.Root, getChildren, p.metrics)
	if err != nil {
		return false, errors.Wrap(err, "correlation.Correlate")
	}
	if err := p.write(ctx, tempDir, groupedBundleData); err != nil {
		return false, err
	}
	savepointID, err := store.Savepoint(ctx)
	if err != nil {
		return false, errors.Wrap(err, "store.Savepoint")
	}
	defer func() {
		if err != nil {
			if rollbackErr := store.RollbackToSavepoint(ctx, savepointID); rollbackErr != nil {
				err = multierror.Append(err, rollbackErr)
			}
		}
	}()
	if err := p.updateXrepoData(ctx, store, upload, groupedBundleData.Packages, groupedBundleData.PackageReferences); err != nil {
		return false, err
	}
	if err := p.sendDB(ctx, upload.ID, tempDir); err != nil {
		return false, err
	}
	return false, nil
}
func ( p * processor ) Process ( ctx context . Context , store store . Store , upload store . Upload ) ( _ bool , err error ) { if cloneInProgress , err := p . isRepoCurrentlyCloning ( ctx , upload . RepositoryID , upload . Commit ) ; err != nil { return false , err } else if cloneInProgress { if err := store . Requeue ( ctx , upload . ID , time . Now ( ) . UTC ( ) . Add ( CloneInProgressDelay ) ) ; err != nil { return false , errors . Wrap ( err , " " ) } return true , nil } tempDir , err := ioutil . TempDir ( " " , " " ) if err != nil { return false , err } defer func ( ) { if cleanupErr := os . RemoveAll ( tempDir ) ; cleanupErr != nil { log15 . Warn ( " " , " " , tempDir , " " , cleanupErr ) } } ( ) r , err := p . bundleManagerClient . GetUpload ( ctx , upload . ID ) if err != nil { return false , errors . Wrap ( err , " " ) } defer func ( ) { if err != nil { if deleteErr := p . bundleManagerClient . DeleteUpload ( ctx , upload . ID ) ; deleteErr != nil { log15 . Warn ( " " , " " , err ) } } } ( ) getChildren := func ( ctx context . Context , dirnames [ ] string ) ( map [ string ] [ ] string , error ) { directoryChildren , err := p . gitserverClient . DirectoryChildren ( ctx , store , upload . RepositoryID , upload . Commit , dirnames ) if err != nil { return nil , errors . Wrap ( err , " " ) } return directoryChildren , nil } groupedBundleData , err := correlation . Correlate ( ctx , r , upload . ID , upload . Root , getChildren , p . metrics ) if err != nil { return false , errors . Wrap ( err , " " ) } if err := p . write ( ctx , tempDir , groupedBundleData ) ; err != nil { return false , err } savepointID , err := store . Savepoint ( ctx ) if err != nil { return false , errors . Wrap ( err , " " ) } defer func ( ) { if err != nil { if rollbackErr := store . RollbackToSavepoint ( ctx , savepointID ) ; rollbackErr != nil { err = multierror . Append ( err , rollbackErr ) } } } ( ) if err := p . updateXrepoData ( ctx , store , upload , groupedBundleData . Packages , groupedBundleData . PackageReferences ) ; err != nil { return false , err } if err := p . sendDB ( ctx , upload . ID , tempDir ) ; err != nil { return false , err } return false , nil }
process converts a raw upload into a dump within the given transaction context.
process converts a raw upload into a dump within the given transaction context .
// Ensure that the repo and revision are resolvable. If the repo does not exist, or if the repo has finished // cloning and the revision does not exist, then the upload will fail to process. If the repo is currently // cloning, then we'll requeue the upload to be tried again later. This will not increase the reset count // of the record (so this doesn't count against the upload as a legitimate attempt). // Create scratch directory that we can clean on completion/failure // Pull raw uploaded data from bundle manager // Remove upload file on error instead of waiting for it to expire // At this point we haven't touched the database. We're going to start a nested transaction // with Postgres savepoints. In the event that something after this point fails, we want to // update the upload record with an error message but do not want to alter any other data in // the database. Rolling back to this savepoint will allow us to discard any other changes // but still commit the transaction as a whole. // Send converted database file to bundle manager