docs: fix typo

Affected endpoint: apiPublishUpdateSwitch (PUT /api/publish/{prefix}/{distribution}).

The handler registered only the published repo key as a task resource.
The underlying source repos (for local) or snapshots (for snapshot-based
published repos) were not locked.  Concurrent updates to a source repo
or snapshot while a publish-update/switch task was running could produce
inconsistent published indexes:

  Task A: apiPublishUpdateSwitch loads published, reads source repo/snapshot
  Request B: modifies same source repo or snapshot (add/remove packages, etc)
  Task A: Update() + Publish() reads stale/modified source -> inconsistent
           published index, or partial write if source deleted mid-task.

Fix: for SourceLocalRepo, iterate published.Sources (component -> source
UUID), look up each local repo via localRepoCollection.ByUUID and append
string(repo.Key()) to resources.  For SourceSnapshot, iterate b.Snapshots,
look up each snapshot via snapshotCollection.ByName and append
string(snapshot.ResourceKey()) to resources.  Task queue now serialises
against both the published repo and all its sources.

CONTRIBUTING.md

@@ -137,7 +137,7 @@ make docker-unit-tests
 
 In order to run aptly system tests, enter the following:
 ```
-make docker-system-tests
+make docker-system-test
 ```
 
 #### Running golangci-lint

Makefile

@@ -241,4 +241,4 @@ clean:  ## remove local build and module cache
 	rm -f unit.out aptly.test VERSION docs/docs.go docs/swagger.json docs/swagger.yaml docs/swagger.conf
 	find system/ -type d -name __pycache__ -exec rm -rf {} \; 2>/dev/null || true
 
-.PHONY: help man prepare swagger version binaries build docker-release docker-system-tests docker-unit-test docker-lint docker-build docker-image docker-man docker-shell docker-serve clean releasetype dpkg serve flake8
+.PHONY: help man prepare swagger version binaries build docker-release docker-system-test docker-unit-test docker-lint docker-build docker-image docker-man docker-shell docker-serve clean releasetype dpkg serve flake8

api/publish.go

@@ -2,6 +2,7 @@ package api
 
 import (
 	"fmt"
+	"log"
 	"net/http"
 	"strings"
 
@@ -298,11 +299,25 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 		multiDist = *b.MultiDist
 	}
 
-	collection := collectionFactory.PublishedRepoCollection()
+	// Pre-register the published repo key in resources so that concurrent
+	// POST requests for the same prefix/distribution are serialized by the
+	// task queue rather than racing on CheckDuplicate + Add.
+	if b.Distribution != "" {
+		storagePrefix := prefix
+		if storage != "" {
+			storagePrefix = storage + ":" + prefix
+		}
+		resources = append(resources, "U"+storagePrefix+">>"+b.Distribution)
+	} else {
+		log.Printf("distribution not specified for publish to prefix '%s' - unable to lock ", prefix)
+	}
 
 	taskName := fmt.Sprintf("Publish %s repository %s/%s with components \"%s\" and sources \"%s\"",
 		b.SourceKind, param, b.Distribution, strings.Join(components, `", "`), strings.Join(names, `", "`))
 	maybeRunTaskInBackground(c, taskName, resources, func(out aptly.Progress, detail *task.Detail) (*task.ProcessReturnValue, error) {
+		taskCollectionFactory := context.NewCollectionFactory()
+		taskCollection := taskCollectionFactory.PublishedRepoCollection()
+
 		taskDetail := task.PublishDetail{
 			Detail: detail,
 		}
@@ -314,10 +329,10 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 		for _, source := range sources {
 			switch s := source.(type) {
 			case *deb.Snapshot:
-				snapshotCollection := collectionFactory.SnapshotCollection()
+				snapshotCollection := taskCollectionFactory.SnapshotCollection()
 				err = snapshotCollection.LoadComplete(s)
 			case *deb.LocalRepo:
-				localCollection := collectionFactory.LocalRepoCollection()
+				localCollection := taskCollectionFactory.LocalRepoCollection()
 				err = localCollection.LoadComplete(s)
 			default:
 				err = fmt.Errorf("unexpected type for source: %T", source)
@@ -327,13 +342,11 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 			}
 		}
 
-		published, err := deb.NewPublishedRepo(storage, prefix, b.Distribution, b.Architectures, components, sources, collectionFactory, multiDist)
+		published, err := deb.NewPublishedRepo(storage, prefix, b.Distribution, b.Architectures, components, sources, taskCollectionFactory, multiDist)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to publish: %s", err)
 		}
 
-		resources = append(resources, string(published.Key()))
-
 		if b.Origin != "" {
 			published.Origin = b.Origin
 		}
@@ -367,18 +380,18 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 			published.Version = b.Version
 		}
 
-		duplicate := collection.CheckDuplicate(published)
+		duplicate := taskCollection.CheckDuplicate(published)
 		if duplicate != nil {
-			_ = collectionFactory.PublishedRepoCollection().LoadComplete(duplicate, collectionFactory)
+			_ = taskCollectionFactory.PublishedRepoCollection().LoadComplete(duplicate, taskCollectionFactory)
 			return &task.ProcessReturnValue{Code: http.StatusBadRequest, Value: nil}, fmt.Errorf("prefix/distribution already used by another published repo: %s", duplicate)
 		}
 
-		err = published.Publish(context.PackagePool(), context, collectionFactory, signer, publishOutput, b.ForceOverwrite, context.SkelPath())
+		err = published.Publish(context.PackagePool(), context, taskCollectionFactory, signer, publishOutput, b.ForceOverwrite, context.SkelPath())
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to publish: %s", err)
 		}
 
-		err = collection.Add(published)
+		err = taskCollection.Add(published)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save to DB: %s", err)
 		}
@@ -458,6 +471,7 @@ func apiPublishUpdateSwitch(c *gin.Context) {
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.PublishedRepoCollection()
 	snapshotCollection := collectionFactory.SnapshotCollection()
+	localRepoCollection := collectionFactory.LocalRepoCollection()
 
 	published, err := collection.ByStoragePrefixDistribution(storage, prefix, distribution)
 	if err != nil {
@@ -465,64 +479,78 @@ func apiPublishUpdateSwitch(c *gin.Context) {
 		return
 	}
 
+	resources := []string{string(published.Key())}
+
 	if published.SourceKind == deb.SourceLocalRepo {
 		if len(b.Snapshots) > 0 {
 			AbortWithJSONError(c, http.StatusBadRequest, fmt.Errorf("snapshots shouldn't be given when updating local repo"))
 			return
 		}
-	} else if published.SourceKind == deb.SourceSnapshot {
-		for _, snapshotInfo := range b.Snapshots {
-			_, err2 := snapshotCollection.ByName(snapshotInfo.Name)
+		for _, uuid := range published.Sources {
+			repo, err2 := localRepoCollection.ByUUID(uuid)
 			if err2 != nil {
 				AbortWithJSONError(c, http.StatusNotFound, err2)
 				return
 			}
+			resources = append(resources, string(repo.Key()))
+		}
+	} else if published.SourceKind == deb.SourceSnapshot {
+		for _, snapshotInfo := range b.Snapshots {
+			snapshot, err2 := snapshotCollection.ByName(snapshotInfo.Name)
+			if err2 != nil {
+				AbortWithJSONError(c, http.StatusNotFound, err2)
+				return
+			}
+			resources = append(resources, string(snapshot.ResourceKey()))
 		}
 	} else {
 		AbortWithJSONError(c, http.StatusInternalServerError, fmt.Errorf("unknown published repository type"))
 		return
 	}
 
-	if b.SkipContents != nil {
-		published.SkipContents = *b.SkipContents
-	}
-
-	if b.SkipBz2 != nil {
-		published.SkipBz2 = *b.SkipBz2
-	}
-
-	if b.AcquireByHash != nil {
-		published.AcquireByHash = *b.AcquireByHash
-	}
-
-	if b.SignedBy != nil {
-		published.SignedBy = *b.SignedBy
-	}
-
-	if b.MultiDist != nil {
-		published.MultiDist = *b.MultiDist
-	}
-
-	if b.Label != nil {
-		published.Label = *b.Label
-	}
-
-	if b.Origin != nil {
-		published.Origin = *b.Origin
-	}
-
-	if b.Version != nil {
-		published.Version = *b.Version
-	}
-
-	resources := []string{string(published.Key())}
+	// Field mutations and fresh DB load are deferred to inside the task so
+	// they always operate on a consistent state after the lock is held.
 	taskName := fmt.Sprintf("Update published %s repository %s/%s", published.SourceKind, published.StoragePrefix(), published.Distribution)
 	maybeRunTaskInBackground(c, taskName, resources, func(out aptly.Progress, _ *task.Detail) (*task.ProcessReturnValue, error) {
-		err = collection.LoadComplete(published, collectionFactory)
+		taskCollectionFactory := context.NewCollectionFactory()
+		taskCollection := taskCollectionFactory.PublishedRepoCollection()
+
+		published, err := taskCollection.ByStoragePrefixDistribution(storage, prefix, distribution)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}
 
+		err = taskCollection.LoadComplete(published, taskCollectionFactory)
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
+		}
+
+		// Apply field mutations on the freshly loaded object.
+		if b.SkipContents != nil {
+			published.SkipContents = *b.SkipContents
+		}
+		if b.SkipBz2 != nil {
+			published.SkipBz2 = *b.SkipBz2
+		}
+		if b.AcquireByHash != nil {
+			published.AcquireByHash = *b.AcquireByHash
+		}
+		if b.SignedBy != nil {
+			published.SignedBy = *b.SignedBy
+		}
+		if b.MultiDist != nil {
+			published.MultiDist = *b.MultiDist
+		}
+		if b.Label != nil {
+			published.Label = *b.Label
+		}
+		if b.Origin != nil {
+			published.Origin = *b.Origin
+		}
+		if b.Version != nil {
+			published.Version = *b.Version
+		}
+
 		revision := published.ObtainRevision()
 		sources := revision.Sources
 
@@ -534,17 +562,17 @@ func apiPublishUpdateSwitch(c *gin.Context) {
 			}
 		}
 
-		result, err := published.Update(collectionFactory, out)
+		result, err := published.Update(taskCollectionFactory, out)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}
 
-		err = published.Publish(context.PackagePool(), context, collectionFactory, signer, out, b.ForceOverwrite, context.SkelPath())
+		err = published.Publish(context.PackagePool(), context, taskCollectionFactory, signer, out, b.ForceOverwrite, context.SkelPath())
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}
 
-		err = collection.Update(published)
+		err = taskCollection.Update(published)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save to DB: %s", err)
 		}
@@ -552,7 +580,7 @@ func apiPublishUpdateSwitch(c *gin.Context) {
 		if b.SkipCleanup == nil || !*b.SkipCleanup {
 			cleanComponents := make([]string, 0, len(result.UpdatedSources)+len(result.RemovedSources))
 			cleanComponents = append(append(cleanComponents, result.UpdatedComponents()...), result.RemovedComponents()...)
-			err = collection.CleanupPrefixComponentFiles(context, published, cleanComponents, collectionFactory, out)
+			err = taskCollection.CleanupPrefixComponentFiles(context, published, cleanComponents, taskCollectionFactory, out)
 			if err != nil {
 				return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 			}
@@ -602,8 +630,11 @@ func apiPublishDrop(c *gin.Context) {
 	resources := []string{string(published.Key())}
 	taskName := fmt.Sprintf("Delete published %s repository %s/%s", published.SourceKind, published.StoragePrefix(), published.Distribution)
 	maybeRunTaskInBackground(c, taskName, resources, func(out aptly.Progress, _ *task.Detail) (*task.ProcessReturnValue, error) {
-		err := collection.Remove(context, storage, prefix, distribution,
-			collectionFactory, out, force, skipCleanup)
+		taskCollectionFactory := context.NewCollectionFactory()
+		taskCollection := taskCollectionFactory.PublishedRepoCollection()
+
+		err := taskCollection.Remove(context, storage, prefix, distribution,
+			taskCollectionFactory, out, force, skipCleanup)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to drop: %s", err)
 		}
@@ -639,43 +670,52 @@ func apiPublishAddSource(c *gin.Context) {
 	storage, prefix := deb.ParsePrefix(param)
 	distribution := slashEscape(c.Params.ByName("distribution"))
 
+	if c.Bind(&b) != nil {
+		return
+	}
+
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.PublishedRepoCollection()
 
+	// Load shallowly (no LoadComplete) to verify existence and obtain the
+	// resource key and task name.  The actual mutation is performed inside
+	// the task on a freshly loaded copy to prevent lost-update races.
 	published, err := collection.ByStoragePrefixDistribution(storage, prefix, distribution)
 	if err != nil {
 		AbortWithJSONError(c, http.StatusNotFound, fmt.Errorf("unable to create: %s", err))
 		return
 	}
 
-	err = collection.LoadComplete(published, collectionFactory)
-	if err != nil {
-		AbortWithJSONError(c, http.StatusInternalServerError, fmt.Errorf("unable to create: %s", err))
-		return
-	}
-
-	if c.Bind(&b) != nil {
-		return
-	}
-
-	revision := published.ObtainRevision()
-	sources := revision.Sources
-
-	component := b.Component
-	name := b.Name
-
-	_, exists := sources[component]
-	if exists {
-		AbortWithJSONError(c, http.StatusBadRequest, fmt.Errorf("unable to create: Component '%s' already exists", component))
-		return
-	}
-
-	sources[component] = name
-
 	resources := []string{string(published.Key())}
 	taskName := fmt.Sprintf("Update published %s repository %s/%s", published.SourceKind, published.StoragePrefix(), published.Distribution)
 	maybeRunTaskInBackground(c, taskName, resources, func(_ aptly.Progress, _ *task.Detail) (*task.ProcessReturnValue, error) {
-		err = collection.Update(published)
+		taskCollectionFactory := context.NewCollectionFactory()
+		taskCollection := taskCollectionFactory.PublishedRepoCollection()
+
+		published, err := taskCollection.ByStoragePrefixDistribution(storage, prefix, distribution)
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusNotFound, Value: nil}, fmt.Errorf("unable to create: %s", err)
+		}
+
+		err = taskCollection.LoadComplete(published, taskCollectionFactory)
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to create: %s", err)
+		}
+
+		revision := published.ObtainRevision()
+		sources := revision.Sources
+
+		component := b.Component
+		name := b.Name
+
+		_, exists := sources[component]
+		if exists {
+			return &task.ProcessReturnValue{Code: http.StatusBadRequest, Value: nil}, fmt.Errorf("unable to create: Component '%s' already exists", component)
+		}
+
+		sources[component] = name
+
+		err = taskCollection.Update(published)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save to DB: %s", err)
 		}
@@ -757,39 +797,48 @@ func apiPublishSetSources(c *gin.Context) {
 	storage, prefix := deb.ParsePrefix(param)
 	distribution := slashEscape(c.Params.ByName("distribution"))
 
+	if c.Bind(&b) != nil {
+		return
+	}
+
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.PublishedRepoCollection()
 
+	// Load shallowly for 404 check, resource key, and task name.
+	// Full load and mutation happen inside the task.
 	published, err := collection.ByStoragePrefixDistribution(storage, prefix, distribution)
 	if err != nil {
 		AbortWithJSONError(c, http.StatusNotFound, fmt.Errorf("unable to update: %s", err))
 		return
 	}
 
-	err = collection.LoadComplete(published, collectionFactory)
-	if err != nil {
-		AbortWithJSONError(c, http.StatusInternalServerError, fmt.Errorf("unable to update: %s", err))
-		return
-	}
-
-	if c.Bind(&b) != nil {
-		return
-	}
-
-	revision := published.ObtainRevision()
-	sources := make(map[string]string, len(b))
-	revision.Sources = sources
-
-	for _, source := range b {
-		component := source.Component
-		name := source.Name
-		sources[component] = name
-	}
-
 	resources := []string{string(published.Key())}
 	taskName := fmt.Sprintf("Update published %s repository %s/%s", published.SourceKind, published.StoragePrefix(), published.Distribution)
 	maybeRunTaskInBackground(c, taskName, resources, func(_ aptly.Progress, _ *task.Detail) (*task.ProcessReturnValue, error) {
-		err = collection.Update(published)
+		taskCollectionFactory := context.NewCollectionFactory()
+		taskCollection := taskCollectionFactory.PublishedRepoCollection()
+
+		published, err := taskCollection.ByStoragePrefixDistribution(storage, prefix, distribution)
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusNotFound, Value: nil}, fmt.Errorf("unable to update: %s", err)
+		}
+
+		err = taskCollection.LoadComplete(published, taskCollectionFactory)
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
+		}
+
+		revision := published.ObtainRevision()
+		sources := make(map[string]string, len(b))
+		revision.Sources = sources
+
+		for _, source := range b {
+			component := source.Component
+			name := source.Name
+			sources[component] = name
+		}
+
+		err = taskCollection.Update(published)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save to DB: %s", err)
 		}
@@ -822,24 +871,33 @@ func apiPublishDropChanges(c *gin.Context) {
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.PublishedRepoCollection()
 
+	// Load shallowly for 404 check, resource key, and task name.
+	// Full load and DropRevision happen inside the task.
 	published, err := collection.ByStoragePrefixDistribution(storage, prefix, distribution)
 	if err != nil {
 		AbortWithJSONError(c, http.StatusNotFound, fmt.Errorf("unable to delete: %s", err))
 		return
 	}
 
-	err = collection.LoadComplete(published, collectionFactory)
-	if err != nil {
-		AbortWithJSONError(c, http.StatusInternalServerError, fmt.Errorf("unable to delete: %s", err))
-		return
-	}
-
-	published.DropRevision()
-
 	resources := []string{string(published.Key())}
 	taskName := fmt.Sprintf("Update published %s repository %s/%s", published.SourceKind, published.StoragePrefix(), published.Distribution)
 	maybeRunTaskInBackground(c, taskName, resources, func(_ aptly.Progress, _ *task.Detail) (*task.ProcessReturnValue, error) {
-		err = collection.Update(published)
+		taskCollectionFactory := context.NewCollectionFactory()
+		taskCollection := taskCollectionFactory.PublishedRepoCollection()
+
+		published, err := taskCollection.ByStoragePrefixDistribution(storage, prefix, distribution)
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusNotFound, Value: nil}, fmt.Errorf("unable to delete: %s", err)
+		}
+
+		err = taskCollection.LoadComplete(published, taskCollectionFactory)
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to delete: %s", err)
+		}
+
+		published.DropRevision()
+
+		err = taskCollection.Update(published)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save to DB: %s", err)
 		}
@@ -875,51 +933,58 @@ func apiPublishUpdateSource(c *gin.Context) {
 	param := slashEscape(c.Params.ByName("prefix"))
 	storage, prefix := deb.ParsePrefix(param)
 	distribution := slashEscape(c.Params.ByName("distribution"))
-	component := slashEscape(c.Params.ByName("component"))
+	urlComponent := slashEscape(c.Params.ByName("component"))
+
+	// Default component to the URL path segment; the body may rename it.
+	b.Component = urlComponent
+	if c.Bind(&b) != nil {
+		return
+	}
 
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.PublishedRepoCollection()
 
+	// Load shallowly for 404 check, resource key, and task name.
+	// Full load and mutation happen inside the task.
 	published, err := collection.ByStoragePrefixDistribution(storage, prefix, distribution)
 	if err != nil {
 		AbortWithJSONError(c, http.StatusNotFound, fmt.Errorf("unable to update: %s", err))
 		return
 	}
 
-	err = collection.LoadComplete(published, collectionFactory)
-	if err != nil {
-		AbortWithJSONError(c, http.StatusInternalServerError, fmt.Errorf("unable to update: %s", err))
-		return
-	}
-
-	revision := published.ObtainRevision()
-	sources := revision.Sources
-
-	_, exists := sources[component]
-	if !exists {
-		AbortWithJSONError(c, http.StatusNotFound, fmt.Errorf("unable to update: Component '%s' does not exist", component))
-		return
-	}
-
-	b.Component = component
-	b.Name = revision.Sources[component]
-
-	if c.Bind(&b) != nil {
-		return
-	}
-
-	if b.Component != component {
-		delete(sources, component)
-	}
-
-	component = b.Component
-	name := b.Name
-	sources[component] = name
-
 	resources := []string{string(published.Key())}
 	taskName := fmt.Sprintf("Update published %s repository %s/%s", published.SourceKind, published.StoragePrefix(), published.Distribution)
 	maybeRunTaskInBackground(c, taskName, resources, func(_ aptly.Progress, _ *task.Detail) (*task.ProcessReturnValue, error) {
-		err = collection.Update(published)
+		taskCollectionFactory := context.NewCollectionFactory()
+		taskCollection := taskCollectionFactory.PublishedRepoCollection()
+
+		published, err := taskCollection.ByStoragePrefixDistribution(storage, prefix, distribution)
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusNotFound, Value: nil}, fmt.Errorf("unable to update: %s", err)
+		}
+
+		err = taskCollection.LoadComplete(published, taskCollectionFactory)
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
+		}
+
+		revision := published.ObtainRevision()
+		sources := revision.Sources
+
+		_, exists := sources[urlComponent]
+		if !exists {
+			return &task.ProcessReturnValue{Code: http.StatusNotFound, Value: nil}, fmt.Errorf("unable to update: Component '%s' does not exist", urlComponent)
+		}
+
+		if b.Component != urlComponent {
+			delete(sources, urlComponent)
+		}
+
+		newComponent := b.Component
+		name := b.Name
+		sources[newComponent] = name
+
+		err = taskCollection.Update(published)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save to DB: %s", err)
 		}
@@ -956,33 +1021,41 @@ func apiPublishRemoveSource(c *gin.Context) {
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.PublishedRepoCollection()
 
+	// Load shallowly for 404 check, resource key, and task name.
+	// Full load and mutation happen inside the task.
 	published, err := collection.ByStoragePrefixDistribution(storage, prefix, distribution)
 	if err != nil {
 		AbortWithJSONError(c, http.StatusNotFound, fmt.Errorf("unable to delete: %s", err))
 		return
 	}
 
-	err = collection.LoadComplete(published, collectionFactory)
-	if err != nil {
-		AbortWithJSONError(c, http.StatusInternalServerError, fmt.Errorf("unable to delete: %s", err))
-		return
-	}
-
-	revision := published.ObtainRevision()
-	sources := revision.Sources
-
-	_, exists := sources[component]
-	if !exists {
-		AbortWithJSONError(c, http.StatusNotFound, fmt.Errorf("unable to delete: Component '%s' does not exist", component))
-		return
-	}
-
-	delete(sources, component)
-
 	resources := []string{string(published.Key())}
 	taskName := fmt.Sprintf("Update published %s repository %s/%s", published.SourceKind, published.StoragePrefix(), published.Distribution)
 	maybeRunTaskInBackground(c, taskName, resources, func(_ aptly.Progress, _ *task.Detail) (*task.ProcessReturnValue, error) {
-		err = collection.Update(published)
+		taskCollectionFactory := context.NewCollectionFactory()
+		taskCollection := taskCollectionFactory.PublishedRepoCollection()
+
+		published, err := taskCollection.ByStoragePrefixDistribution(storage, prefix, distribution)
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusNotFound, Value: nil}, fmt.Errorf("unable to delete: %s", err)
+		}
+
+		err = taskCollection.LoadComplete(published, taskCollectionFactory)
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to delete: %s", err)
+		}
+
+		revision := published.ObtainRevision()
+		sources := revision.Sources
+
+		_, exists := sources[component]
+		if !exists {
+			return &task.ProcessReturnValue{Code: http.StatusNotFound, Value: nil}, fmt.Errorf("unable to delete: Component '%s' does not exist", component)
+		}
+
+		delete(sources, component)
+
+		err = taskCollection.Update(published)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save to DB: %s", err)
 		}
@@ -1054,64 +1127,94 @@ func apiPublishUpdate(c *gin.Context) {
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.PublishedRepoCollection()
 
+	// Load shallowly for 404 check, resource key, and task name.
+	// Full load and field mutations happen inside the task.
 	published, err := collection.ByStoragePrefixDistribution(storage, prefix, distribution)
 	if err != nil {
 		AbortWithJSONError(c, http.StatusNotFound, fmt.Errorf("unable to update: %s", err))
 		return
 	}
 
-	err = collection.LoadComplete(published, collectionFactory)
-	if err != nil {
-		AbortWithJSONError(c, http.StatusInternalServerError, fmt.Errorf("unable to update: %s", err))
-		return
-	}
-
-	if b.SkipContents != nil {
-		published.SkipContents = *b.SkipContents
-	}
-
-	if b.SkipBz2 != nil {
-		published.SkipBz2 = *b.SkipBz2
-	}
-
-	if b.AcquireByHash != nil {
-		published.AcquireByHash = *b.AcquireByHash
-	}
-
-	if b.SignedBy != nil {
-		published.SignedBy = *b.SignedBy
-	}
-
-	if b.MultiDist != nil {
-		published.MultiDist = *b.MultiDist
-	}
-
-	if b.Label != nil {
-		published.Label = *b.Label
-	}
-
-	if b.Origin != nil {
-		published.Origin = *b.Origin
-	}
-
-	if b.Version != nil {
-		published.Version = *b.Version
-	}
-
 	resources := []string{string(published.Key())}
+
+	// Lock source repos / snapshots the same way apiPublishUpdateSwitch does,
+	// because published.Update() reads from them and concurrent modification
+	// would produce an inconsistent view.
+	snapshotCollection := collectionFactory.SnapshotCollection()
+	localRepoCollection := collectionFactory.LocalRepoCollection()
+
+	if published.SourceKind == deb.SourceLocalRepo {
+		for _, uuid := range published.Sources {
+			repo, err2 := localRepoCollection.ByUUID(uuid)
+			if err2 != nil {
+				AbortWithJSONError(c, http.StatusNotFound, err2)
+				return
+			}
+			resources = append(resources, string(repo.Key()))
+		}
+	} else if published.SourceKind == deb.SourceSnapshot {
+		for _, uuid := range published.Sources {
+			snapshot, err2 := snapshotCollection.ByUUID(uuid)
+			if err2 != nil {
+				AbortWithJSONError(c, http.StatusNotFound, err2)
+				return
+			}
+			resources = append(resources, string(snapshot.ResourceKey()))
+		}
+	}
+
 	taskName := fmt.Sprintf("Update published %s repository %s/%s", published.SourceKind, published.StoragePrefix(), published.Distribution)
 	maybeRunTaskInBackground(c, taskName, resources, func(out aptly.Progress, _ *task.Detail) (*task.ProcessReturnValue, error) {
-		result, err := published.Update(collectionFactory, out)
+		taskCollectionFactory := context.NewCollectionFactory()
+		taskCollection := taskCollectionFactory.PublishedRepoCollection()
+
+		published, err := taskCollection.ByStoragePrefixDistribution(storage, prefix, distribution)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}
 
-		err = published.Publish(context.PackagePool(), context, collectionFactory, signer, out, b.ForceOverwrite, context.SkelPath())
+		err = taskCollection.LoadComplete(published, taskCollectionFactory)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}
 
-		err = collection.Update(published)
+		// Apply field mutations on the freshly loaded object.
+		if b.SkipContents != nil {
+			published.SkipContents = *b.SkipContents
+		}
+		if b.SkipBz2 != nil {
+			published.SkipBz2 = *b.SkipBz2
+		}
+		if b.AcquireByHash != nil {
+			published.AcquireByHash = *b.AcquireByHash
+		}
+		if b.SignedBy != nil {
+			published.SignedBy = *b.SignedBy
+		}
+		if b.MultiDist != nil {
+			published.MultiDist = *b.MultiDist
+		}
+		if b.Label != nil {
+			published.Label = *b.Label
+		}
+		if b.Origin != nil {
+			published.Origin = *b.Origin
+		}
+		if b.Version != nil {
+			published.Version = *b.Version
+		}
+
+		result, err := published.Update(taskCollectionFactory, out)
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
+		}
+
+		err = published.Publish(context.PackagePool(), context, taskCollectionFactory, signer, out, b.ForceOverwrite, context.SkelPath())
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
+		}
+
+		err = taskCollection.Update(published)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save to DB: %s", err)
 		}
@@ -1119,7 +1222,7 @@ func apiPublishUpdate(c *gin.Context) {
 		if b.SkipCleanup == nil || !*b.SkipCleanup {
 			cleanComponents := make([]string, 0, len(result.UpdatedSources)+len(result.RemovedSources))
 			cleanComponents = append(append(cleanComponents, result.UpdatedComponents()...), result.RemovedComponents()...)
-			err = collection.CleanupPrefixComponentFiles(context, published, cleanComponents, collectionFactory, out)
+			err = taskCollection.CleanupPrefixComponentFiles(context, published, cleanComponents, taskCollectionFactory, out)
 			if err != nil {
 				return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 			}