Fix concurrent publish operations causing missing package files

When multiple repository operations execute concurrently on shared pool
directories, race conditions could cause .deb files to be deleted despite
appearing in repository metadata, resulting in apt 404 errors.

Three distinct but related race conditions were identified and fixed:

1. Package addition vs publish race: When packages are added to a local
   repository that is already published, the publish operation could read
   stale package references before the add transaction commits. Fixed by
   locking all published repositories that reference the local repo during
   package addition.

2. Pool file deletion race: When multiple published repositories share the
   same pool directory (same storage+prefix) and publish concurrently, cleanup
   operations could delete each other's newly created files. The cleanup in
   thread B would:
   - Query database for referenced files (not seeing thread A's uncommitted files)
   - Scan pool directory (seeing thread A's files)
   - Delete thread A's files as "orphaned"

   Fixed by implementing pool-sibling locking: acquire locks on ALL published
   repositories sharing the same storage and prefix before publish/cleanup.

3. Concurrent cleanup on same prefix: Multiple distributions publishing to the
   same prefix concurrently could have cleanup operations delete shared files.
   Fixed by:
   - Adding prefix-level locking to serialize cleanup operations
   - Removing ref subtraction that incorrectly marked shared files as orphaned
   - Forcing database reload before cleanup to see recent commits

The existing task system serializes operations based on resource locks,
preventing these race conditions when proper lock sets are acquired.

Test coverage includes concurrent publish scenarios that reliably reproduced
all three bugs before the fixes.

Author: Agustin Henze

api/publish.go

@@ -465,7 +465,18 @@ func apiPublishUpdateSwitch(c *gin.Context) {
 		published.MultiDist = *b.MultiDist
 	}
 
-	resources := []string{string(published.Key())}
+	// Lock the pool prefix and all published repos sharing it to prevent cleanup race conditions.
+	// Without prefix-level locking, concurrent publishes to different distributions under the
+	// same prefix can race during cleanup, causing one operation to delete the other's files.
+	poolLockKey := fmt.Sprintf("pool:%s:%s", storage, prefix)
+	resources := []string{poolLockKey, string(published.Key())}
+	poolSiblings := collection.ByStoragePrefix(storage, prefix)
+	for _, sibling := range poolSiblings {
+		if sibling.UUID != published.UUID {
+			resources = append(resources, string(sibling.Key()))
+		}
+	}
+
 	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)
@@ -1024,7 +1035,18 @@ func apiPublishUpdate(c *gin.Context) {
 		published.MultiDist = *b.MultiDist
 	}
 
-	resources := []string{string(published.Key())}
+	// Lock the pool prefix and all published repos sharing it to prevent cleanup race conditions.
+	// Without prefix-level locking, concurrent publishes to different distributions under the
+	// same prefix can race during cleanup, causing one operation to delete the other's files.
+	poolLockKey := fmt.Sprintf("pool:%s:%s", storage, prefix)
+	resources := []string{poolLockKey, string(published.Key())}
+	poolSiblings := collection.ByStoragePrefix(storage, prefix)
+	for _, sibling := range poolSiblings {
+		if sibling.UUID != published.UUID {
+			resources = append(resources, string(sibling.Key()))
+		}
+	}
+
 	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)