Add task api and resource locking ability

api/api.go

@@ -8,6 +8,7 @@ import (
 	"github.com/aptly-dev/aptly/aptly"
 	"github.com/aptly-dev/aptly/deb"
 	"github.com/aptly-dev/aptly/query"
+	"github.com/aptly-dev/aptly/task"
 	"github.com/gin-gonic/gin"
 )
 
@@ -34,12 +35,14 @@ type dbRequest struct {
 	err  chan<- error
 }
 
+var dbRequests chan dbRequest
+
 // Acquire database lock and release it when not needed anymore.
 //
 // Should be run in a goroutine!
-func acquireDatabase(requests <-chan dbRequest) {
+func acquireDatabase() {
 	clients := 0
-	for request := range requests {
+	for request := range dbRequests {
 		var err error
 
 		switch request.kind {
@@ -66,6 +69,46 @@ func acquireDatabase(requests <-chan dbRequest) {
 	}
 }
 
+// Should be called before database access is needed in any api call.
+// Happens per default for each api call. It is important that you run
+// runTaskInBackground to run a task which accquire database.
+// Important do not forget to defer to releaseDatabaseConnection
+func acquireDatabaseConnection() error {
+	if dbRequests == nil {
+		return nil
+	}
+
+	errCh := make(chan error)
+	dbRequests <- dbRequest{acquiredb, errCh}
+
+	return <-errCh
+}
+
+// Release database connection when not needed anymore
+func releaseDatabaseConnection() error {
+	if dbRequests == nil {
+		return nil
+	}
+
+	errCh := make(chan error)
+	dbRequests <- dbRequest{releasedb, errCh}
+	return <-errCh
+}
+
+// runs tasks in background. Acquires database connection first.
+func runTaskInBackground(name string, resources []string, proc task.Process) (task.Task, *task.ResourceConflictError) {
+	return context.TaskList().RunTaskInBackground(name, resources, func(out *task.Output, detail *task.Detail) error {
+		err := acquireDatabaseConnection()
+
+		if err != nil {
+			return err
+		}
+
+		defer releaseDatabaseConnection()
+		return proc(out, detail)
+	})
+}
+
 // Common piece of code to show list of packages,
 // with searching & details if requested
 func showPackages(c *gin.Context, reflist *deb.PackageRefList, collectionFactory *deb.CollectionFactory) {

api/publish.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/aptly-dev/aptly/deb"
 	"github.com/aptly-dev/aptly/pgp"
+	"github.com/aptly-dev/aptly/task"
 	"github.com/aptly-dev/aptly/utils"
 	"github.com/gin-gonic/gin"
 )
@@ -114,7 +115,9 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 	}
 
 	var components []string
+	var names []string
 	var sources []interface{}
+	var resources []string
 	collectionFactory := context.NewCollectionFactory()
 
 	if b.SourceKind == "snapshot" {
@@ -124,6 +127,7 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 
 		for _, source := range b.Sources {
 			components = append(components, source.Component)
+			names = append(names, source.Name)
 
 			snapshot, err = snapshotCollection.ByName(source.Name)
 			if err != nil {
@@ -131,6 +135,7 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 				return
 			}
 
+			resources = append(resources, string(snapshot.ResourceKey()))
 			err = snapshotCollection.LoadComplete(snapshot)
 			if err != nil {
 				c.AbortWithError(500, fmt.Errorf("unable to publish: %s", err))
@@ -146,6 +151,7 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 
 		for _, source := range b.Sources {
 			components = append(components, source.Component)
+			names = append(names, source.Name)
 
 			localRepo, err = localCollection.ByName(source.Name)
 			if err != nil {
@@ -153,6 +159,7 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 				return
 			}
 
+			resources = append(resources, string(localRepo.Key()))
 			err = localCollection.LoadComplete(localRepo)
 			if err != nil {
 				c.AbortWithError(500, fmt.Errorf("unable to publish: %s", err))
@@ -165,53 +172,62 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 		return
 	}
 
-	collection := collectionFactory.PublishedRepoCollection()
-
 	published, err := deb.NewPublishedRepo(storage, prefix, b.Distribution, b.Architectures, components, sources, collectionFactory)
 	if err != nil {
 		c.AbortWithError(500, fmt.Errorf("unable to publish: %s", err))
 		return
 	}
-	if b.Origin != "" {
-		published.Origin = b.Origin
-	}
-	if b.NotAutomatic != "" {
-		published.NotAutomatic = b.NotAutomatic
-	}
-	if b.ButAutomaticUpgrades != "" {
-		published.ButAutomaticUpgrades = b.ButAutomaticUpgrades
-	}
-	published.Label = b.Label
 
-	published.SkipContents = context.Config().SkipContentsPublishing
-	if b.SkipContents != nil {
-		published.SkipContents = *b.SkipContents
-	}
+	resources = append(resources, string(published.Key()))
+	collection := collectionFactory.PublishedRepoCollection()
 
-	if b.AcquireByHash != nil {
-		published.AcquireByHash = *b.AcquireByHash
-	}
+	taskName := fmt.Sprintf("Publish %s: %s", b.SourceKind, strings.Join(names, ", "))
+	task, conflictErr := runTaskInBackground(taskName, resources, func(out *task.Output, detail *task.Detail) error {
+		if b.Origin != "" {
+			published.Origin = b.Origin
+		}
+		if b.NotAutomatic != "" {
+			published.NotAutomatic = b.NotAutomatic
+		}
+		if b.ButAutomaticUpgrades != "" {
+			published.ButAutomaticUpgrades = b.ButAutomaticUpgrades
+		}
+		published.Label = b.Label
 
-	duplicate := collection.CheckDuplicate(published)
-	if duplicate != nil {
-		collectionFactory.PublishedRepoCollection().LoadComplete(duplicate, collectionFactory)
-		c.AbortWithError(400, fmt.Errorf("prefix/distribution already used by another published repo: %s", duplicate))
+		published.SkipContents = context.Config().SkipContentsPublishing
+		if b.SkipContents != nil {
+			published.SkipContents = *b.SkipContents
+		}
+
+		if b.AcquireByHash != nil {
+			published.AcquireByHash = *b.AcquireByHash
+		}
+
+		duplicate := collection.CheckDuplicate(published)
+		if duplicate != nil {
+			collectionFactory.PublishedRepoCollection().LoadComplete(duplicate, collectionFactory)
+			return fmt.Errorf("prefix/distribution already used by another published repo: %s", duplicate)
+		}
+
+		err := published.Publish(context.PackagePool(), context, collectionFactory, signer, out, b.ForceOverwrite)
+		if err != nil {
+			return fmt.Errorf("unable to publish: %s", err)
+		}
+
+		err = collection.Add(published)
+		if err != nil {
+			return fmt.Errorf("unable to save to DB: %s", err)
+		}
+
+		return nil
+	})
+
+	if conflictErr != nil {
+		c.AbortWithError(409, conflictErr)
 		return
 	}
 
-	err = published.Publish(context.PackagePool(), context, collectionFactory, signer, nil, b.ForceOverwrite)
-	if err != nil {
-		c.AbortWithError(500, fmt.Errorf("unable to publish: %s", err))
-		return
-	}
-
-	err = collection.Add(published)
-	if err != nil {
-		c.AbortWithError(500, fmt.Errorf("unable to save to DB: %s", err))
-		return
-	}
-
-	c.JSON(201, published)
+	c.JSON(202, task)
 }
 
 // PUT /publish/:prefix/:distribution
@@ -257,6 +273,8 @@ func apiPublishUpdateSwitch(c *gin.Context) {
 	}
 
 	var updatedComponents []string
+	var updatedSnapshots []string
+	var resources []string
 
 	if published.SourceKind == deb.SourceLocalRepo {
 		if len(b.Snapshots) > 0 {
@@ -290,6 +308,7 @@ func apiPublishUpdateSwitch(c *gin.Context) {
 
 			published.UpdateSnapshot(snapshotInfo.Component, snapshot)
 			updatedComponents = append(updatedComponents, snapshotInfo.Component)
+			updatedSnapshots = append(updatedSnapshots, snapshot.Name)
 		}
 	} else {
 		c.AbortWithError(500, fmt.Errorf("unknown published repository type"))
@@ -304,28 +323,36 @@ func apiPublishUpdateSwitch(c *gin.Context) {
 		published.AcquireByHash = *b.AcquireByHash
 	}
 
-	err = published.Publish(context.PackagePool(), context, collectionFactory, signer, nil, b.ForceOverwrite)
-	if err != nil {
-		c.AbortWithError(500, fmt.Errorf("unable to update: %s", err))
-		return
-	}
-
-	err = collection.Update(published)
-	if err != nil {
-		c.AbortWithError(500, fmt.Errorf("unable to save to DB: %s", err))
-		return
-	}
-
-	if b.SkipCleanup == nil || !*b.SkipCleanup {
-		err = collection.CleanupPrefixComponentFiles(published.Prefix, updatedComponents,
-			context.GetPublishedStorage(storage), collectionFactory, nil)
+	resources = append(resources, string(published.Key()))
+	taskName := fmt.Sprintf("Update published %s (%s): %s", published.SourceKind, strings.Join(updatedComponents, " "), strings.Join(updatedSnapshots, ", "))
+	currTask, conflictErr := runTaskInBackground(taskName, resources, func(out *task.Output, detail *task.Detail) error {
+		err := published.Publish(context.PackagePool(), context, collectionFactory, signer, out, b.ForceOverwrite)
 		if err != nil {
-			c.AbortWithError(500, fmt.Errorf("unable to update: %s", err))
-			return
+			return fmt.Errorf("unable to update: %s", err)
 		}
+
+		err = collection.Update(published)
+		if err != nil {
+			return fmt.Errorf("unable to save to DB: %s", err)
+		}
+
+		if b.SkipCleanup == nil || !*b.SkipCleanup {
+			err = collection.CleanupPrefixComponentFiles(published.Prefix, updatedComponents,
+				context.GetPublishedStorage(storage), collectionFactory, out)
+			if err != nil {
+				return fmt.Errorf("unable to update: %s", err)
+			}
+		}
+
+		return nil
+	})
+
+	if conflictErr != nil {
+		c.AbortWithError(409, conflictErr)
+		return
 	}
 
-	c.JSON(200, published)
+	c.JSON(202, currTask)
 }
 
 // DELETE /publish/:prefix/:distribution
@@ -340,12 +367,29 @@ func apiPublishDrop(c *gin.Context) {
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.PublishedRepoCollection()
 
-	err := collection.Remove(context, storage, prefix, distribution,
-		collectionFactory, context.Progress(), force, skipCleanup)
+	published, err := collection.ByStoragePrefixDistribution(storage, prefix, distribution)
 	if err != nil {
 		c.AbortWithError(500, fmt.Errorf("unable to drop: %s", err))
 		return
 	}
 
-	c.JSON(200, gin.H{})
+	resources := []string{string(published.Key())}
+
+	taskName := fmt.Sprintf("Delete published %s (%s)", prefix, distribution)
+	currTask, conflictErr := runTaskInBackground(taskName, resources, func(out *task.Output, detail *task.Detail) error {
+		err := collection.Remove(context, storage, prefix, distribution,
+			collectionFactory, out, force, skipCleanup)
+		if err != nil {
+			return fmt.Errorf("unable to drop: %s", err)
+		}
+
+		return nil
+	})
+
+	if conflictErr != nil {
+		c.AbortWithError(409, conflictErr)
+		return
+	}
+
+	c.JSON(202, currTask)
 }

api/repos.go

@@ -4,11 +4,14 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"strings"
+	"text/template"
 
 	"github.com/aptly-dev/aptly/aptly"
 	"github.com/aptly-dev/aptly/database"
 	"github.com/aptly-dev/aptly/deb"
 	"github.com/aptly-dev/aptly/query"
+	"github.com/aptly-dev/aptly/task"
 	"github.com/aptly-dev/aptly/utils"
 	"github.com/gin-gonic/gin"
 )
@@ -112,39 +115,43 @@ func apiReposShow(c *gin.Context) {
 // DELETE /api/repos/:name
 func apiReposDrop(c *gin.Context) {
 	force := c.Request.URL.Query().Get("force") == "1"
+	name := c.Params.ByName("name")
 
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.LocalRepoCollection()
 	snapshotCollection := collectionFactory.SnapshotCollection()
 	publishedCollection := collectionFactory.PublishedRepoCollection()
 
-	repo, err := collection.ByName(c.Params.ByName("name"))
+	repo, err := collection.ByName(name)
 	if err != nil {
 		c.AbortWithError(404, err)
 		return
 	}
 
-	published := publishedCollection.ByLocalRepo(repo)
-	if len(published) > 0 {
-		c.AbortWithError(409, fmt.Errorf("unable to drop, local repo is published"))
-		return
-	}
-
-	if !force {
-		snapshots := snapshotCollection.ByLocalRepoSource(repo)
-		if len(snapshots) > 0 {
-			c.AbortWithError(409, fmt.Errorf("unable to drop, local repo has snapshots, use ?force=1 to override"))
-			return
+	resources := []string{string(repo.Key())}
+	taskName := fmt.Sprintf("Delete repo %s", name)
+	task, conflictErr := runTaskInBackground(taskName, resources, func(out *task.Output, detail *task.Detail) error {
+		published := publishedCollection.ByLocalRepo(repo)
+		if len(published) > 0 {
+			return fmt.Errorf("unable to drop, local repo is published")
 		}
-	}
 
-	err = collection.Drop(repo)
-	if err != nil {
-		c.AbortWithError(500, err)
+		if !force {
+			snapshots := snapshotCollection.ByLocalRepoSource(repo)
+			if len(snapshots) > 0 {
+				return fmt.Errorf("unable to drop, local repo has snapshots, use ?force=1 to override")
+			}
+		}
+
+		return collection.Drop(repo)
+	})
+
+	if conflictErr != nil {
+		c.AbortWithError(409, conflictErr)
 		return
 	}
 
-	c.JSON(200, gin.H{})
+	c.JSON(202, task)
 }
 
 // GET /api/repos/:name/packages
@@ -168,7 +175,7 @@ func apiReposPackagesShow(c *gin.Context) {
 }
 
 // Handler for both add and delete
-func apiReposPackagesAddDelete(c *gin.Context, cb func(list *deb.PackageList, p *deb.Package) error) {
+func apiReposPackagesAddDelete(c *gin.Context, taskNamePrefix string, cb func(list *deb.PackageList, p *deb.Package, out *task.Output) error) {
 	var b struct {
 		PackageRefs []string
 	}
@@ -192,54 +199,57 @@ func apiReposPackagesAddDelete(c *gin.Context, cb func(list *deb.PackageList, p
 		return
 	}
 
-	list, err := deb.NewPackageListFromRefList(repo.RefList(), collectionFactory.PackageCollection(), nil)
-	if err != nil {
-		c.AbortWithError(500, err)
-		return
-	}
-
-	// verify package refs and build package list
-	for _, ref := range b.PackageRefs {
-		var p *deb.Package
-
-		p, err = collectionFactory.PackageCollection().ByKey([]byte(ref))
+	resources := []string{string(repo.Key())}
+	currTask, conflictErr := runTaskInBackground(taskNamePrefix+repo.Name, resources, func(out *task.Output, detail *task.Detail) error {
+		out.Print("Loading packages...\n")
+		list, err := deb.NewPackageListFromRefList(repo.RefList(), collectionFactory.PackageCollection(), nil)
 		if err != nil {
-			if err == database.ErrNotFound {
-				c.AbortWithError(404, fmt.Errorf("package %s: %s", ref, err))
-			} else {
-				c.AbortWithError(500, err)
+			return err
+		}
+
+		// verify package refs and build package list
+		for _, ref := range b.PackageRefs {
+			var p *deb.Package
+
+			p, err = collectionFactory.PackageCollection().ByKey([]byte(ref))
+			if err != nil {
+				if err == database.ErrNotFound {
+					return fmt.Errorf("packages %s: %s", ref, err)
+				}
+
+				return err
+			}
+			err = cb(list, p, out)
+			if err != nil {
+				return err
 			}
-			return
 		}
-		err = cb(list, p)
-		if err != nil {
-			c.AbortWithError(400, err)
-			return
-		}
-	}
 
-	repo.UpdateRefList(deb.NewPackageRefListFromPackageList(list))
+		repo.UpdateRefList(deb.NewPackageRefListFromPackageList(list))
 
-	err = collectionFactory.LocalRepoCollection().Update(repo)
-	if err != nil {
-		c.AbortWithError(500, fmt.Errorf("unable to save: %s", err))
+		return collectionFactory.LocalRepoCollection().Update(repo)
+	})
+
+	if conflictErr != nil {
+		c.AbortWithError(409, conflictErr)
 		return
 	}
 
-	c.JSON(200, repo)
-
+	c.JSON(202, currTask)
 }
 
 // POST /repos/:name/packages
 func apiReposPackagesAdd(c *gin.Context) {
-	apiReposPackagesAddDelete(c, func(list *deb.PackageList, p *deb.Package) error {
+	apiReposPackagesAddDelete(c, "Add packages to repo ", func(list *deb.PackageList, p *deb.Package, out *task.Output) error {
+		out.Printf("Adding package %s\n", p.Name)
 		return list.Add(p)
 	})
 }
 
 // DELETE /repos/:name/packages
 func apiReposPackagesDelete(c *gin.Context) {
-	apiReposPackagesAddDelete(c, func(list *deb.PackageList, p *deb.Package) error {
+	apiReposPackagesAddDelete(c, "Delete packages from repo ", func(list *deb.PackageList, p *deb.Package, out *task.Output) error {
+		out.Printf("Removing package %s\n", p.Name)
 		list.Remove(p)
 		return nil
 	})
@@ -260,6 +270,7 @@ func apiReposPackageFromDir(c *gin.Context) {
 		return
 	}
 
+	dirParam := c.Params.ByName("dir")
 	fileParam := c.Params.ByName("file")
 	if fileParam != "" && !verifyPath(fileParam) {
 		c.AbortWithError(400, fmt.Errorf("wrong file"))
@@ -269,7 +280,8 @@ func apiReposPackageFromDir(c *gin.Context) {
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.LocalRepoCollection()
 
-	repo, err := collection.ByName(c.Params.ByName("name"))
+	name := c.Params.ByName("name")
+	repo, err := collection.ByName(name)
 	if err != nil {
 		c.AbortWithError(404, err)
 		return
@@ -281,76 +293,96 @@ func apiReposPackageFromDir(c *gin.Context) {
 		return
 	}
 
-	verifier := context.GetVerifier()
-
-	var (
-		sources                      []string
-		packageFiles, failedFiles    []string
-		otherFiles                   []string
-		processedFiles, failedFiles2 []string
-		reporter                     = &aptly.RecordingResultReporter{
-			Warnings:     []string{},
-			AddedLines:   []string{},
-			RemovedLines: []string{},
-		}
-		list *deb.PackageList
-	)
-
+	var taskName string
+	var sources []string
 	if fileParam == "" {
-		sources = []string{filepath.Join(context.UploadPath(), c.Params.ByName("dir"))}
+		taskName = fmt.Sprintf("Add packages from dir %s to repo %s", dirParam, name)
+		sources = []string{filepath.Join(context.UploadPath(), dirParam)}
 	} else {
-		sources = []string{filepath.Join(context.UploadPath(), c.Params.ByName("dir"), c.Params.ByName("file"))}
+		sources = []string{filepath.Join(context.UploadPath(), dirParam, fileParam)}
+		taskName = fmt.Sprintf("Add package %s from dir %s to repo %s", fileParam, dirParam, name)
 	}
 
-	packageFiles, otherFiles, failedFiles = deb.CollectPackageFiles(sources, reporter)
+	resources := []string{string(repo.Key())}
+	resources = append(resources, sources...)
+	currTask, conflictErr := runTaskInBackground(taskName, resources, func(out *task.Output, detail *task.Detail) error {
+		verifier := context.GetVerifier()
 
-	list, err = deb.NewPackageListFromRefList(repo.RefList(), collectionFactory.PackageCollection(), nil)
-	if err != nil {
-		c.AbortWithError(500, fmt.Errorf("unable to load packages: %s", err))
-		return
-	}
-
-	processedFiles, failedFiles2, err = deb.ImportPackageFiles(list, packageFiles, forceReplace, verifier, context.PackagePool(),
-		collectionFactory.PackageCollection(), reporter, nil, collectionFactory.ChecksumCollection)
-	failedFiles = append(failedFiles, failedFiles2...)
-
-	processedFiles = append(processedFiles, otherFiles...)
-
-	if err != nil {
-		c.AbortWithError(500, fmt.Errorf("unable to import package files: %s", err))
-		return
-	}
-
-	repo.UpdateRefList(deb.NewPackageRefListFromPackageList(list))
-
-	err = collectionFactory.LocalRepoCollection().Update(repo)
-	if err != nil {
-		c.AbortWithError(500, fmt.Errorf("unable to save: %s", err))
-		return
-	}
-
-	if !noRemove {
-		processedFiles = utils.StrSliceDeduplicate(processedFiles)
-
-		for _, file := range processedFiles {
-			err := os.Remove(file)
-			if err != nil {
-				reporter.Warning("unable to remove file %s: %s", file, err)
+		var (
+			packageFiles, failedFiles    []string
+			otherFiles                   []string
+			processedFiles, failedFiles2 []string
+			reporter                     = &aptly.RecordingResultReporter{
+				Warnings:     []string{},
+				AddedLines:   []string{},
+				RemovedLines: []string{},
 			}
+			list *deb.PackageList
+		)
+
+		packageFiles, otherFiles, failedFiles = deb.CollectPackageFiles(sources, reporter)
+
+		list, err := deb.NewPackageListFromRefList(repo.RefList(), collectionFactory.PackageCollection(), nil)
+		if err != nil {
+			return fmt.Errorf("unable to load packages: %s", err)
 		}
 
-		// atempt to remove dir, if it fails, that's fine: probably it's not empty
-		os.Remove(filepath.Join(context.UploadPath(), c.Params.ByName("dir")))
-	}
+		processedFiles, failedFiles2, err = deb.ImportPackageFiles(list, packageFiles, forceReplace, verifier, context.PackagePool(),
+			collectionFactory.PackageCollection(), reporter, nil, collectionFactory.ChecksumCollection)
+		failedFiles = append(failedFiles, failedFiles2...)
+		processedFiles = append(processedFiles, otherFiles...)
 
-	if failedFiles == nil {
-		failedFiles = []string{}
-	}
+		if err != nil {
+			return fmt.Errorf("unable to import package files: %s", err)
+		}
 
-	c.JSON(200, gin.H{
-		"Report":      reporter,
-		"FailedFiles": failedFiles,
+		repo.UpdateRefList(deb.NewPackageRefListFromPackageList(list))
+
+		err = collectionFactory.LocalRepoCollection().Update(repo)
+		if err != nil {
+			return fmt.Errorf("unable to save: %s", err)
+		}
+
+		if !noRemove {
+			processedFiles = utils.StrSliceDeduplicate(processedFiles)
+
+			for _, file := range processedFiles {
+				err := os.Remove(file)
+				if err != nil {
+					reporter.Warning("unable to remove file %s: %s", file, err)
+				}
+			}
+
+			// atempt to remove dir, if it fails, that's fine: probably it's not empty
+			os.Remove(filepath.Join(context.UploadPath(), dirParam))
+		}
+
+		if failedFiles == nil {
+			failedFiles = []string{}
+		}
+
+		if len(reporter.AddedLines) > 0 {
+			out.Printf("Added: %s\n", strings.Join(reporter.AddedLines, ", "))
+		}
+		if len(reporter.RemovedLines) > 0 {
+			out.Printf("Removed: %s\n", strings.Join(reporter.RemovedLines, ", "))
+		}
+		if len(reporter.Warnings) > 0 {
+			out.Printf("Warnings: %s\n", strings.Join(reporter.Warnings, ", "))
+		}
+		if len(failedFiles) > 0 {
+			out.Printf("Failed files: %s\n", strings.Join(failedFiles, ", "))
+		}
+
+		return nil
 	})
+
+	if conflictErr != nil {
+		c.AbortWithError(409, conflictErr)
+		return
+	}
+
+	c.JSON(202, currTask)
 }
 
 // POST /repos/:name/include/:dir/:file
@@ -367,59 +399,103 @@ func apiReposIncludePackageFromDir(c *gin.Context) {
 	ignoreSignature := c.Request.URL.Query().Get("ignoreSignature") == "1"
 
 	repoTemplateString := c.Params.ByName("name")
+	collectionFactory := context.NewCollectionFactory()
 
 	if !verifyDir(c) {
 		return
 	}
 
+	var sources []string
+	var taskName string
+	dirParam := c.Params.ByName("dir")
 	fileParam := c.Params.ByName("file")
 	if fileParam != "" && !verifyPath(fileParam) {
 		c.AbortWithError(400, fmt.Errorf("wrong file"))
 		return
 	}
 
-	var (
-		err                       error
-		verifier                  = context.GetVerifier()
-		sources, changesFiles     []string
-		failedFiles, failedFiles2 []string
-		reporter                  = &aptly.RecordingResultReporter{
-			Warnings:     []string{},
-			AddedLines:   []string{},
-			RemovedLines: []string{},
-		}
-	)
-
 	if fileParam == "" {
-		sources = []string{filepath.Join(context.UploadPath(), c.Params.ByName("dir"))}
+		taskName = fmt.Sprintf("Include packages from changes files in dir %s to repo matching template %s", dirParam, repoTemplateString)
+		sources = []string{filepath.Join(context.UploadPath(), dirParam)}
 	} else {
-		sources = []string{filepath.Join(context.UploadPath(), c.Params.ByName("dir"), c.Params.ByName("file"))}
+		taskName = fmt.Sprintf("Include packages from changes file %s from dir %s to repo matching template %s", fileParam, dirParam, repoTemplateString)
+		sources = []string{filepath.Join(context.UploadPath(), dirParam, fileParam)}
 	}
 
-	collectionFactory := context.NewCollectionFactory()
-	changesFiles, failedFiles = deb.CollectChangesFiles(sources, reporter)
-	_, failedFiles2, err = deb.ImportChangesFiles(
-		changesFiles, reporter, acceptUnsigned, ignoreSignature, forceReplace, noRemoveFiles, verifier,
-		repoTemplateString, context.Progress(), collectionFactory.LocalRepoCollection(), collectionFactory.PackageCollection(),
-		context.PackagePool(), collectionFactory.ChecksumCollection, nil, query.Parse)
-	failedFiles = append(failedFiles, failedFiles2...)
-
+	repoTemplate, err := template.New("repo").Parse(repoTemplateString)
 	if err != nil {
-		c.AbortWithError(500, fmt.Errorf("unable to import changes files: %s", err))
+		c.AbortWithError(400, fmt.Errorf("error parsing repo template: %s", err))
 		return
 	}
 
-	if !noRemoveFiles {
-		// atempt to remove dir, if it fails, that's fine: probably it's not empty
-		os.Remove(filepath.Join(context.UploadPath(), c.Params.ByName("dir")))
-	}
+	var resources []string
+	if len(repoTemplate.Tree.Root.Nodes) > 1 {
+		resources = append(resources, task.AllLocalReposResourcesKey)
+	} else {
+		// repo template string is simple text so only use resource key of specific repository
+		repo, err := collectionFactory.LocalRepoCollection().ByName(repoTemplateString)
+		if err != nil {
+			c.AbortWithError(404, err)
+			return
+		}
 
-	if failedFiles == nil {
-		failedFiles = []string{}
+		resources = append(resources, string(repo.Key()))
 	}
+	resources = append(resources, sources...)
 
-	c.JSON(200, gin.H{
-		"Report":      reporter,
-		"FailedFiles": failedFiles,
+	currTask, conflictErr := runTaskInBackground(taskName, resources, func(out *task.Output, detail *task.Detail) error {
+		var (
+			err                       error
+			verifier                  = context.GetVerifier()
+			changesFiles              []string
+			failedFiles, failedFiles2 []string
+			reporter                  = &aptly.RecordingResultReporter{
+				Warnings:     []string{},
+				AddedLines:   []string{},
+				RemovedLines: []string{},
+			}
+		)
+
+		changesFiles, failedFiles = deb.CollectChangesFiles(sources, reporter)
+		_, failedFiles2, err = deb.ImportChangesFiles(
+			changesFiles, reporter, acceptUnsigned, ignoreSignature, forceReplace, noRemoveFiles, verifier,
+			repoTemplate, context.Progress(), collectionFactory.LocalRepoCollection(), collectionFactory.PackageCollection(),
+			context.PackagePool(), collectionFactory.ChecksumCollection, nil, query.Parse)
+		failedFiles = append(failedFiles, failedFiles2...)
+
+		if err != nil {
+			return fmt.Errorf("unable to import changes files: %s", err)
+		}
+
+		if !noRemoveFiles {
+			// atempt to remove dir, if it fails, that's fine: probably it's not empty
+			os.Remove(filepath.Join(context.UploadPath(), dirParam))
+		}
+
+		if failedFiles == nil {
+			failedFiles = []string{}
+		}
+
+		if len(reporter.AddedLines) > 0 {
+			out.Printf("Added: %s\n", strings.Join(reporter.AddedLines, ", "))
+		}
+		if len(reporter.RemovedLines) > 0 {
+			out.Printf("Removed: %s\n", strings.Join(reporter.RemovedLines, ", "))
+		}
+		if len(reporter.Warnings) > 0 {
+			out.Printf("Warnings: %s\n", strings.Join(reporter.Warnings, ", "))
+		}
+		if len(failedFiles) > 0 {
+			out.Printf("Failed files: %s\n", strings.Join(failedFiles, ", "))
+		}
+
+		return nil
 	})
+
+	if conflictErr != nil {
+		c.AbortWithError(409, conflictErr)
+		return
+	}
+
+	c.JSON(202, currTask)
 }

api/router.go

@@ -20,15 +20,15 @@ func Router(c *ctx.AptlyContext) http.Handler {
 		// We use a goroutine to count the number of
 		// concurrent requests. When no more requests are
 		// running, we close the database to free the lock.
-		requests := make(chan dbRequest)
+		dbRequests = make(chan dbRequest)
 
-		go acquireDatabase(requests)
+		go acquireDatabase()
 
 		router.Use(func(c *gin.Context) {
 			var err error
 
 			errCh := make(chan error)
-			requests <- dbRequest{acquiredb, errCh}
+			dbRequests <- dbRequest{acquiredb, errCh}
 
 			err = <-errCh
 			if err != nil {
@@ -37,7 +37,7 @@ func Router(c *ctx.AptlyContext) http.Handler {
 			}
 
 			defer func() {
-				requests <- dbRequest{releasedb, errCh}
+				dbRequests <- dbRequest{releasedb, errCh}
 				err = <-errCh
 				if err != nil {
 					c.AbortWithError(500, err)
@@ -111,6 +111,16 @@ func Router(c *ctx.AptlyContext) http.Handler {
 	{
 		root.GET("/graph.:ext", apiGraph)
 	}
+	{
+		root.GET("/tasks", apiTasksList)
+		root.POST("/tasks-clear", apiTasksClear)
+		root.GET("/tasks-wait", apiTasksWait)
+		root.GET("/tasks/:id/wait", apiTasksWaitForTaskByID)
+		root.GET("/tasks/:id/output", apiTasksOutputShow)
+		root.GET("/tasks/:id/detail", apiTasksDetailShow)
+		root.GET("/tasks/:id", apiTasksShow)
+		root.DELETE("/tasks/:id", apiTasksDelete)
+	}
 
 	return router
 }

api/snapshot.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/aptly-dev/aptly/database"
 	"github.com/aptly-dev/aptly/deb"
+	"github.com/aptly-dev/aptly/task"
 	"github.com/gin-gonic/gin"
 )
 
@@ -48,42 +49,46 @@ func apiSnapshotsCreateFromMirror(c *gin.Context) {
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.RemoteRepoCollection()
 	snapshotCollection := collectionFactory.SnapshotCollection()
+	name := c.Params.ByName("name")
 
-	repo, err = collection.ByName(c.Params.ByName("name"))
+	repo, err = collection.ByName(name)
 	if err != nil {
 		c.AbortWithError(404, err)
 		return
 	}
 
-	err = repo.CheckLock()
-	if err != nil {
-		c.AbortWithError(409, err)
+	// including snapshot resource key
+	resources := []string{string(repo.Key()), "S" + b.Name}
+	taskName := fmt.Sprintf("Create snapshot of mirror %s", name)
+	currTask, conflictErr := runTaskInBackground(taskName, resources, func(out *task.Output, detail *task.Detail) error {
+		err := repo.CheckLock()
+		if err != nil {
+			return err
+		}
+
+		err = collection.LoadComplete(repo)
+		if err != nil {
+			return err
+		}
+
+		snapshot, err = deb.NewSnapshotFromRepository(b.Name, repo)
+		if err != nil {
+			return err
+		}
+
+		if b.Description != "" {
+			snapshot.Description = b.Description
+		}
+
+		return snapshotCollection.Add(snapshot)
+	})
+
+	if conflictErr != nil {
+		c.AbortWithError(409, conflictErr)
 		return
 	}
 
-	err = collection.LoadComplete(repo)
-	if err != nil {
-		c.AbortWithError(500, err)
-		return
-	}
-
-	snapshot, err = deb.NewSnapshotFromRepository(b.Name, repo)
-	if err != nil {
-		c.AbortWithError(400, err)
-		return
-	}
-
-	if b.Description != "" {
-		snapshot.Description = b.Description
-	}
-
-	err = snapshotCollection.Add(snapshot)
-	if err != nil {
-		c.AbortWithError(400, err)
-		return
-	}
-
-	c.JSON(201, snapshot)
+	c.JSON(202, currTask)
 }
 
 // POST /api/snapshots
@@ -112,6 +117,7 @@ func apiSnapshotsCreate(c *gin.Context) {
 
 	collectionFactory := context.NewCollectionFactory()
 	snapshotCollection := collectionFactory.SnapshotCollection()
+	var resources []string
 
 	sources := make([]*deb.Snapshot, len(b.SourceSnapshots))
 
@@ -127,39 +133,39 @@ func apiSnapshotsCreate(c *gin.Context) {
 			c.AbortWithError(500, err)
 			return
 		}
+
+		resources = append(resources, string(sources[i].ResourceKey()))
 	}
 
-	list := deb.NewPackageList()
+	currTask, conflictErr := runTaskInBackground("Create snapshot "+b.Name, resources, func(out *task.Output, detail *task.Detail) error {
+		list := deb.NewPackageList()
 
-	// verify package refs and build package list
-	for _, ref := range b.PackageRefs {
-		var p *deb.Package
-
-		p, err = collectionFactory.PackageCollection().ByKey([]byte(ref))
-		if err != nil {
-			if err == database.ErrNotFound {
-				c.AbortWithError(404, fmt.Errorf("package %s: %s", ref, err))
-			} else {
-				c.AbortWithError(500, err)
+		// verify package refs and build package list
+		for _, ref := range b.PackageRefs {
+			p, err := collectionFactory.PackageCollection().ByKey([]byte(ref))
+			if err != nil {
+				if err == database.ErrNotFound {
+					return fmt.Errorf("package %s: %s", ref, err)
+				}
+				return err
+			}
+			err = list.Add(p)
+			if err != nil {
+				return err
 			}
-			return
 		}
-		err = list.Add(p)
-		if err != nil {
-			c.AbortWithError(400, err)
-			return
-		}
-	}
 
-	snapshot = deb.NewSnapshotFromRefList(b.Name, sources, deb.NewPackageRefListFromPackageList(list), b.Description)
+		snapshot = deb.NewSnapshotFromRefList(b.Name, sources, deb.NewPackageRefListFromPackageList(list), b.Description)
 
-	err = snapshotCollection.Add(snapshot)
-	if err != nil {
-		c.AbortWithError(400, err)
+		return snapshotCollection.Add(snapshot)
+	})
+
+	if conflictErr != nil {
+		c.AbortWithError(409, conflictErr)
 		return
 	}
 
-	c.JSON(201, snapshot)
+	c.JSON(202, currTask)
 }
 
 // POST /api/repos/:name/snapshots
@@ -182,36 +188,41 @@ func apiSnapshotsCreateFromRepository(c *gin.Context) {
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.LocalRepoCollection()
 	snapshotCollection := collectionFactory.SnapshotCollection()
+	name := c.Params.ByName("name")
 
-	repo, err = collection.ByName(c.Params.ByName("name"))
+	repo, err = collection.ByName(name)
 	if err != nil {
 		c.AbortWithError(404, err)
 		return
 	}
 
-	err = collection.LoadComplete(repo)
-	if err != nil {
-		c.AbortWithError(500, err)
+	// including snapshot resource key
+	resources := []string{string(repo.Key()), "S" + b.Name}
+	taskName := fmt.Sprintf("Create snapshot of repo %s", name)
+	currTask, conflictErr := runTaskInBackground(taskName, resources, func(out *task.Output, detail *task.Detail) error {
+		err := collection.LoadComplete(repo)
+		if err != nil {
+			return err
+		}
+
+		snapshot, err = deb.NewSnapshotFromLocalRepo(b.Name, repo)
+		if err != nil {
+			return err
+		}
+
+		if b.Description != "" {
+			snapshot.Description = b.Description
+		}
+
+		return snapshotCollection.Add(snapshot)
+	})
+
+	if conflictErr != nil {
+		c.AbortWithError(409, conflictErr)
 		return
 	}
 
-	snapshot, err = deb.NewSnapshotFromLocalRepo(b.Name, repo)
-	if err != nil {
-		c.AbortWithError(400, err)
-		return
-	}
-
-	if b.Description != "" {
-		snapshot.Description = b.Description
-	}
-
-	err = snapshotCollection.Add(snapshot)
-	if err != nil {
-		c.AbortWithError(400, err)
-		return
-	}
-
-	c.JSON(201, snapshot)
+	c.JSON(202, currTask)
 }
 
 // PUT /api/snapshots/:name
@@ -232,34 +243,39 @@ func apiSnapshotsUpdate(c *gin.Context) {
 
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.SnapshotCollection()
+	name := c.Params.ByName("name")
 
-	snapshot, err = collection.ByName(c.Params.ByName("name"))
+	snapshot, err = collection.ByName(name)
 	if err != nil {
 		c.AbortWithError(404, err)
 		return
 	}
 
-	_, err = collection.ByName(b.Name)
-	if err == nil {
-		c.AbortWithError(409, fmt.Errorf("unable to rename: snapshot %s already exists", b.Name))
+	resources := []string{string(snapshot.ResourceKey()), "S" + b.Name}
+	taskName := fmt.Sprintf("Update snapshot %s", name)
+	currTask, conflictErr := runTaskInBackground(taskName, resources, func(out *task.Output, detail *task.Detail) error {
+		_, err := collection.ByName(b.Name)
+		if err == nil {
+			return fmt.Errorf("unable to rename: snapshot %s already exists", b.Name)
+		}
+
+		if b.Name != "" {
+			snapshot.Name = b.Name
+		}
+
+		if b.Description != "" {
+			snapshot.Description = b.Description
+		}
+
+		return collectionFactory.SnapshotCollection().Update(snapshot)
+	})
+
+	if conflictErr != nil {
+		c.AbortWithError(409, conflictErr)
 		return
 	}
 
-	if b.Name != "" {
-		snapshot.Name = b.Name
-	}
-
-	if b.Description != "" {
-		snapshot.Description = b.Description
-	}
-
-	err = collectionFactory.SnapshotCollection().Update(snapshot)
-	if err != nil {
-		c.AbortWithError(500, err)
-		return
-	}
-
-	c.JSON(200, snapshot)
+	c.JSON(202, currTask)
 }
 
 // GET /api/snapshots/:name
@@ -297,28 +313,31 @@ func apiSnapshotsDrop(c *gin.Context) {
 		return
 	}
 
-	published := publishedCollection.BySnapshot(snapshot)
+	resources := []string{string(snapshot.ResourceKey())}
+	taskName := fmt.Sprintf("Delete snapshot %s", name)
+	currTask, conflictErr := runTaskInBackground(taskName, resources, func(out *task.Output, detail *task.Detail) error {
+		published := publishedCollection.BySnapshot(snapshot)
 
-	if len(published) > 0 {
-		c.AbortWithError(409, fmt.Errorf("unable to drop: snapshot is published"))
-		return
-	}
-
-	if !force {
-		snapshots := snapshotCollection.BySnapshotSource(snapshot)
-		if len(snapshots) > 0 {
-			c.AbortWithError(409, fmt.Errorf("won't delete snapshot that was used as source for other snapshots, use ?force=1 to override"))
-			return
+		if len(published) > 0 {
+			return fmt.Errorf("unable to drop: snapshot is published")
 		}
-	}
 
-	err = snapshotCollection.Drop(snapshot)
-	if err != nil {
-		c.AbortWithError(500, err)
+		if !force {
+			snapshots := snapshotCollection.BySnapshotSource(snapshot)
+			if len(snapshots) > 0 {
+				return fmt.Errorf("won't delete snapshot that was used as source for other snapshots, use ?force=1 to override")
+			}
+		}
+
+		return snapshotCollection.Drop(snapshot)
+	})
+
+	if conflictErr != nil {
+		c.AbortWithError(409, conflictErr)
 		return
 	}
 
-	c.JSON(200, gin.H{})
+	c.JSON(202, currTask)
 }
 
 // GET /api/snapshots/:name/diff/:withSnapshot

api/task.go

@@ -0,0 +1,122 @@
+package api
+
+import (
+	"strconv"
+
+	"github.com/aptly-dev/aptly/task"
+	"github.com/gin-gonic/gin"
+)
+
+// GET /tasks
+func apiTasksList(c *gin.Context) {
+	list := context.TaskList()
+	c.JSON(200, list.GetTasks())
+}
+
+// POST /tasks/clear
+func apiTasksClear(c *gin.Context) {
+	list := context.TaskList()
+	list.Clear()
+	c.JSON(200, gin.H{})
+}
+
+// GET /tasks-wait
+func apiTasksWait(c *gin.Context) {
+	list := context.TaskList()
+	list.Wait()
+	c.JSON(200, gin.H{})
+}
+
+// GET /tasks/:id/wait
+func apiTasksWaitForTaskByID(c *gin.Context) {
+	list := context.TaskList()
+	id, err := strconv.ParseInt(c.Params.ByName("id"), 10, 0)
+	if err != nil {
+		c.AbortWithError(500, err)
+		return
+	}
+
+	task, err := list.WaitForTaskByID(int(id))
+	if err != nil {
+		c.AbortWithError(400, err)
+		return
+	}
+
+	c.JSON(200, task)
+}
+
+// GET /tasks/:id
+func apiTasksShow(c *gin.Context) {
+	list := context.TaskList()
+	id, err := strconv.ParseInt(c.Params.ByName("id"), 10, 0)
+	if err != nil {
+		c.AbortWithError(500, err)
+		return
+	}
+
+	var task task.Task
+	task, err = list.GetTaskByID(int(id))
+	if err != nil {
+		c.AbortWithError(404, err)
+		return
+	}
+
+	c.JSON(200, task)
+}
+
+// GET /tasks/:id/output
+func apiTasksOutputShow(c *gin.Context) {
+	list := context.TaskList()
+	id, err := strconv.ParseInt(c.Params.ByName("id"), 10, 0)
+	if err != nil {
+		c.AbortWithError(500, err)
+		return
+	}
+
+	var output string
+	output, err = list.GetTaskOutputByID(int(id))
+	if err != nil {
+		c.AbortWithError(404, err)
+		return
+	}
+
+	c.JSON(200, output)
+}
+
+// GET /tasks/:id/detail
+func apiTasksDetailShow(c *gin.Context) {
+	list := context.TaskList()
+	id, err := strconv.ParseInt(c.Params.ByName("id"), 10, 0)
+	if err != nil {
+		c.AbortWithError(500, err)
+		return
+	}
+
+	var detail interface{}
+	detail, err = list.GetTaskDetailByID(int(id))
+	if err != nil {
+		c.AbortWithError(404, err)
+		return
+	}
+
+	c.JSON(200, detail)
+}
+
+// DELETE /tasks/:id
+func apiTasksDelete(c *gin.Context) {
+	list := context.TaskList()
+	id, err := strconv.ParseInt(c.Params.ByName("id"), 10, 0)
+	if err != nil {
+		c.AbortWithError(500, err)
+		return
+	}
+
+	var delTask task.Task
+	delTask, err = list.DeleteTaskByID(int(id))
+	if err != nil {
+		c.AbortWithError(400, err)
+		return
+	}
+
+	c.JSON(200, delTask)
+}