Fix/task race

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

@@ -182,6 +182,9 @@ binaries: prepare swagger  ## Build binary releases (FreeBSD, macOS, Linux gener
 docker-image:  ## Build aptly-dev docker image
 	@docker build -f system/Dockerfile . -t aptly-dev
 
+docker-image-test:  # Build aptly-test docker image for testing
+	@docker build -f docker/test.Dockerfile . -t aptly-test
+
 docker-image-no-cache:  ## Build aptly-dev docker image (no cache)
 	@docker build --no-cache -f system/Dockerfile . -t aptly-dev
 
@@ -241,4 +244,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/mirror.go

@@ -216,9 +216,9 @@ func apiMirrorsDrop(c *gin.Context) {
 	name := c.Params.ByName("name")
 	force := c.Request.URL.Query().Get("force") == "1"
 
+	// Phase 1: Pre-task validation (shallow load for 404 check only)
 	collectionFactory := context.NewCollectionFactory()
 	mirrorCollection := collectionFactory.RemoteRepoCollection()
-	snapshotCollection := collectionFactory.SnapshotCollection()
 
 	repo, err := mirrorCollection.ByName(name)
 	if err != nil {
@@ -228,21 +228,34 @@ func apiMirrorsDrop(c *gin.Context) {
 
 	resources := []string{string(repo.Key())}
 	taskName := fmt.Sprintf("Delete mirror %s", name)
+
 	maybeRunTaskInBackground(c, taskName, resources, func(_ aptly.Progress, _ *task.Detail) (*task.ProcessReturnValue, error) {
-		err := repo.CheckLock()
+		// Phase 2: Inside task lock - create fresh collections
+		taskCollectionFactory := context.NewCollectionFactory()
+		taskMirrorCollection := taskCollectionFactory.RemoteRepoCollection()
+		taskSnapshotCollection := taskCollectionFactory.SnapshotCollection()
+
+		// Fresh load after lock acquired
+		repo, err := taskMirrorCollection.ByName(name)
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to drop: %v", err)
+		}
+
+		err = repo.CheckLock()
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to drop: %v", err)
 		}
 
 		if !force {
-			snapshots := snapshotCollection.ByRemoteRepoSource(repo)
+			// Fresh checks with current collections
+			snapshots := taskSnapshotCollection.ByRemoteRepoSource(repo)
 
 			if len(snapshots) > 0 {
 				return &task.ProcessReturnValue{Code: http.StatusForbidden, Value: nil}, fmt.Errorf("won't delete mirror with snapshots, use 'force=1' to override")
 			}
 		}
 
-		err = mirrorCollection.Drop(repo)
+		err = taskMirrorCollection.Drop(repo)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to drop: %v", err)
 		}
@@ -535,7 +548,8 @@ func apiMirrorsUpdate(c *gin.Context) {
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.RemoteRepoCollection()
 
-	remote, err = collection.ByName(c.Params.ByName("name"))
+	name := c.Params.ByName("name")
+	remote, err = collection.ByName(name)
 	if err != nil {
 		AbortWithJSONError(c, 404, err)
 		return
@@ -550,6 +564,7 @@ func apiMirrorsUpdate(c *gin.Context) {
 		return
 	}
 
+	// Pre-task validation of new name if provided
 	if b.Name != remote.Name {
 		_, err = collection.ByName(b.Name)
 		if err == nil {
@@ -566,9 +581,26 @@ func apiMirrorsUpdate(c *gin.Context) {
 
 	resources := []string{string(remote.Key())}
 	maybeRunTaskInBackground(c, "Update mirror "+b.Name, resources, func(out aptly.Progress, detail *task.Detail) (*task.ProcessReturnValue, error) {
+		// Phase 2: Inside task lock - create fresh factory
+		taskCollectionFactory := context.NewCollectionFactory()
+		taskCollection := taskCollectionFactory.RemoteRepoCollection()
+
+		// Fresh load after lock acquired (use captured `name` variable, not gin context)
+		remote, err := taskCollection.ByName(name)
+		if err != nil {
+			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
+		}
+
+		// Fresh rename check inside lock (if renaming)
+		if b.Name != remote.Name {
+			_, err := taskCollection.ByName(b.Name)
+			if err == nil {
+				return &task.ProcessReturnValue{Code: http.StatusConflict, Value: nil}, fmt.Errorf("unable to rename: mirror %s already exists", b.Name)
+			}
+		}
 
 		downloader := context.NewDownloader(out)
-		err := remote.Fetch(downloader, verifier, b.IgnoreSignatures)
+		err = remote.Fetch(downloader, verifier, b.IgnoreSignatures)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}
@@ -580,14 +612,14 @@ func apiMirrorsUpdate(c *gin.Context) {
 			}
 		}
 
-		err = remote.DownloadPackageIndexes(out, downloader, verifier, collectionFactory, b.IgnoreSignatures, remote.SkipComponentCheck)
+		err = remote.DownloadPackageIndexes(out, downloader, verifier, taskCollectionFactory, b.IgnoreSignatures, remote.SkipComponentCheck)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}
 
 		if remote.DownloadAppStream && !remote.IsFlat() {
 			err = remote.DownloadAppStreamFiles(out, downloader,
-				context.PackagePool(), collectionFactory.ChecksumCollection(nil), b.IgnoreChecksums)
+				context.PackagePool(), taskCollectionFactory.ChecksumCollection(nil), b.IgnoreChecksums)
 			if err != nil {
 				return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 			}
@@ -607,8 +639,8 @@ func apiMirrorsUpdate(c *gin.Context) {
 			}
 		}
 
-		queue, downloadSize, err := remote.BuildDownloadQueue(context.PackagePool(), collectionFactory.PackageCollection(),
-			collectionFactory.ChecksumCollection(nil), b.SkipExistingPackages, b.LatestOnly)
+		queue, downloadSize, err := remote.BuildDownloadQueue(context.PackagePool(), taskCollectionFactory.PackageCollection(),
+			taskCollectionFactory.ChecksumCollection(nil), b.SkipExistingPackages, b.LatestOnly)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}
@@ -618,12 +650,12 @@ func apiMirrorsUpdate(c *gin.Context) {
 			e := context.ReOpenDatabase()
 			if e == nil {
 				remote.MarkAsIdle()
-				_ = collection.Update(remote)
+				_ = taskCollection.Update(remote)
 			}
 		}()
 
 		remote.MarkAsUpdating()
-		err = collection.Update(remote)
+		err = taskCollection.Update(remote)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}
@@ -727,7 +759,7 @@ func apiMirrorsUpdate(c *gin.Context) {
 						}
 
 						// and import it back to the pool
-						task.File.PoolPath, err = context.PackagePool().Import(task.TempDownPath, task.File.Filename, &task.File.Checksums, true, collectionFactory.ChecksumCollection(nil))
+						task.File.PoolPath, err = context.PackagePool().Import(task.TempDownPath, task.File.Filename, &task.File.Checksums, true, taskCollectionFactory.ChecksumCollection(nil))
 						if err != nil {
 							//return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to import file: %s", err)
 							pushError(err)
@@ -780,8 +812,8 @@ func apiMirrorsUpdate(c *gin.Context) {
 		}
 
 		log.Info().Msgf("%s: Finalizing download...", b.Name)
-		_ = remote.FinalizeDownload(collectionFactory, out)
-		err = collectionFactory.RemoteRepoCollection().Update(remote)
+		_ = remote.FinalizeDownload(taskCollectionFactory, out)
+		err = taskCollection.Update(remote)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}