deb: make the index rename concurrency configurable

publishRenameConcurrency (publish_rename_concurrency in YAML) sets how
many index files RenameFiles renames at once. It defaults to 4, the
value that was hard-coded, and 1 renames them one at a time. Values
below 1 are treated as 1, since a semaphore with no capacity would
never admit a rename.

The setting is global rather than per publishing endpoint because the
renames run against every storage backend, not only S3.

deb/index_files.go

@@ -500,13 +500,6 @@ func (files *indexFiles) FinalizeAll(progress aptly.Progress, signer pgp.Signer)
 	return
 }
 
-// How many index renames are in flight at once. Every rename is a round trip,
-// and on an S3 backend it is a CopyObject followed by a DeleteObject, so a
-// publish rewriting a dozen indexes spent most of its wall clock waiting on
-// them one at a time. Four matches the default used for concurrent package
-// uploads.
-const renameConcurrency = 4
-
 func (files *indexFiles) RenameFiles() error {
 	type renamePair struct{ oldName, newName string }
 
@@ -520,7 +513,8 @@ func (files *indexFiles) RenameFiles() error {
 		mu       sync.Mutex
 		firstErr error
 	)
-	sem := make(chan struct{}, renameConcurrency)
+	// Below 1 the semaphore would admit nothing, so treat it as serial.
+	sem := make(chan struct{}, max(utils.Config.PublishRenameConcurrency, 1))
 
 	for _, pair := range pairs {
 		wg.Add(1)

deb/index_files_rename_test.go

@@ -3,16 +3,28 @@ package deb
 import (
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/aptly-dev/aptly/aptly"
+	"github.com/aptly-dev/aptly/utils"
 
 	. "gopkg.in/check.v1"
 )
 
-type RenameFilesSuite struct{}
+type RenameFilesSuite struct {
+	savedRenameConcurrency int
+}
 
 var _ = Suite(&RenameFilesSuite{})
 
+func (s *RenameFilesSuite) SetUpTest(c *C) {
+	s.savedRenameConcurrency = utils.Config.PublishRenameConcurrency
+}
+
+func (s *RenameFilesSuite) TearDownTest(c *C) {
+	utils.Config.PublishRenameConcurrency = s.savedRenameConcurrency
+}
+
 // Only RenameFile is exercised by RenameFiles, so the rest of the interface is
 // embedded and left nil: a call to anything else would panic, which is the
 // loudest way to notice this fake drifting from what the code under test does.
@@ -24,6 +36,12 @@ type renameRecordingStorage struct {
 	failOn   string
 	inFlight int
 	maxSeen  int
+
+	// With waitFor set, every rename blocks until that many run at once, so a
+	// test proves the limit is reached rather than only respected.
+	waitFor  int
+	released bool
+	release  chan struct{}
 }
 
 func (s *renameRecordingStorage) RenameFile(oldName, newName string) error {
@@ -32,6 +50,10 @@ func (s *renameRecordingStorage) RenameFile(oldName, newName string) error {
 	if s.inFlight > s.maxSeen {
 		s.maxSeen = s.inFlight
 	}
+	if s.waitFor > 0 && s.inFlight == s.waitFor && !s.released {
+		s.released = true
+		close(s.release)
+	}
 	s.mu.Unlock()
 
 	defer func() {
@@ -40,6 +62,14 @@ func (s *renameRecordingStorage) RenameFile(oldName, newName string) error {
 		s.mu.Unlock()
 	}()
 
+	if s.waitFor > 0 {
+		select {
+		case <-s.release:
+		case <-time.After(5 * time.Second):
+			return fmt.Errorf("never saw %d renames in flight at once", s.waitFor)
+		}
+	}
+
 	if oldName == s.failOn {
 		return fmt.Errorf("rename refused for %s", oldName)
 	}
@@ -53,7 +83,7 @@ func (s *renameRecordingStorage) RenameFile(oldName, newName string) error {
 func newRenameFiles(storage aptly.PublishedStorage, n int) (*indexFiles, *renameRecordingStorage) {
 	rec, ok := storage.(*renameRecordingStorage)
 	if !ok {
-		rec = &renameRecordingStorage{renamed: make(map[string]string)}
+		rec = &renameRecordingStorage{renamed: make(map[string]string), release: make(chan struct{})}
 	}
 	files := &indexFiles{
 		publishedStorage: rec,
@@ -77,15 +107,28 @@ func (s *RenameFilesSuite) TestRenameFilesRenamesEveryEntry(c *C) {
 	}
 }
 
-func (s *RenameFilesSuite) TestRenameFilesStaysWithinTheConcurrencyLimit(c *C) {
+func (s *RenameFilesSuite) TestRenameFilesRunsUpToTheConfiguredLimit(c *C) {
+	// Above the default of 4, so a limit that ignored the config would time out.
+	utils.Config.PublishRenameConcurrency = 6
 	files, rec := newRenameFiles(nil, 50)
+	rec.waitFor = 6
 
 	c.Assert(files.RenameFiles(), IsNil)
 
-	// An upper bound, so this cannot flake: fewer in flight than the limit is
-	// always acceptable, more never is.
-	c.Check(rec.maxSeen <= renameConcurrency, Equals, true,
-		Commentf("saw %d renames in flight, limit is %d", rec.maxSeen, renameConcurrency))
+	c.Check(rec.maxSeen, Equals, 6)
+	c.Check(len(rec.renamed), Equals, 50)
+}
+
+func (s *RenameFilesSuite) TestRenameFilesIsSerialAtALimitOfOneOrLess(c *C) {
+	for _, limit := range []int{1, 0, -3} {
+		utils.Config.PublishRenameConcurrency = limit
+		files, rec := newRenameFiles(nil, 10)
+
+		c.Assert(files.RenameFiles(), IsNil, Commentf("limit %d", limit))
+
+		c.Check(rec.maxSeen, Equals, 1, Commentf("limit %d", limit))
+		c.Check(len(rec.renamed), Equals, 10, Commentf("limit %d", limit))
+	}
 }
 
 func (s *RenameFilesSuite) TestRenameFilesReportsAFailure(c *C) {

debian/aptly.conf

@@ -171,6 +171,10 @@ skip_contents_publishing: false
 # Do not create bz2 files
 skip_bz2_publishing: false
 
+# Number of index files renamed in parallel when a publish finishes
+# (set to 1 to rename them one at a time)
+publish_rename_concurrency: 4
+
 
 # Storage
 ##########

man/aptly.1

@@ -201,6 +201,10 @@ The legacy json configuration is still supported (and also supports comments):
   // Do not create bz2 files
   "skipBz2Publishing": false,
 
+  // Number of index files renamed in parallel when a publish finishes
+  // (set to 1 to rename them one at a time)
+  "publishRenameConcurrency": 4,
+
 
 // Storage
 ///////////

man/aptly.1.ronn.tmpl

@@ -190,6 +190,10 @@ The legacy json configuration is still supported (and also supports comments):
       // Do not create bz2 files
       "skipBz2Publishing": false,
 
+      // Number of index files renamed in parallel when a publish finishes
+      // (set to 1 to rename them one at a time)
+      "publishRenameConcurrency": 4,
+
 
     // Storage
     ///////////

system/t02_config/ConfigShowTest_gold

@@ -40,6 +40,7 @@
     "gpgKeys": [],
     "skipContentsPublishing": false,
     "skipBz2Publishing": false,
+    "publishRenameConcurrency": 4,
     "FileSystemPublishEndpoints": {},
     "JFrogPublishEndpoints": null,
     "S3PublishEndpoints": {},

system/t02_config/ConfigShowYAMLTest_gold

@@ -37,6 +37,7 @@ gpg_disable_verify: false
 gpg_keys: []
 skip_contents_publishing: false
 skip_bz2_publishing: false
+publish_rename_concurrency: 4
 filesystem_publish_endpoints: {}
 jfrog_publish_endpoints: {}
 s3_publish_endpoints: {}

system/t02_config/CreateConfigTest_gold

@@ -171,6 +171,10 @@ skip_contents_publishing: false
 # Do not create bz2 files
 skip_bz2_publishing: false
 
+# Number of index files renamed in parallel when a publish finishes
+# (set to 1 to rename them one at a time)
+publish_rename_concurrency: 4
+
 
 # Storage
 ##########

utils/config.go

@@ -62,8 +62,9 @@ type ConfigStructure struct { // nolint: maligned
 	GpgKeys          []string `json:"gpgKeys"                       yaml:"gpg_keys"`
 
 	// Publishing
-	SkipContentsPublishing bool `json:"skipContentsPublishing"        yaml:"skip_contents_publishing"`
-	SkipBz2Publishing      bool `json:"skipBz2Publishing"             yaml:"skip_bz2_publishing"`
+	SkipContentsPublishing   bool `json:"skipContentsPublishing"        yaml:"skip_contents_publishing"`
+	SkipBz2Publishing        bool `json:"skipBz2Publishing"             yaml:"skip_bz2_publishing"`
+	PublishRenameConcurrency int  `json:"publishRenameConcurrency"      yaml:"publish_rename_concurrency"`
 
 	// Storage
 	FileSystemPublishRoots map[string]FileSystemPublishRoot `json:"FileSystemPublishEndpoints"    yaml:"filesystem_publish_endpoints"`
@@ -254,6 +255,7 @@ var Config = ConfigStructure{
 	RootDir:                filepath.Join(os.Getenv("HOME"), ".aptly"),
 	AcquireByHashAlgorithms: []string{"MD5Sum", "SHA1", "SHA256", "SHA512"},
 	DownloadConcurrency:    4,
+	PublishRenameConcurrency: 4,
 	DownloadLimit:          0,
 	Downloader:             "default",
 	DatabaseOpenAttempts:   -1,

utils/config_test.go

@@ -27,6 +27,16 @@ func (s *ConfigSuite) TestLoadConfig(c *C) {
 	c.Check(s.config.GetRootDir(), Equals, "/opt/aptly/")
 	c.Check(s.config.DownloadConcurrency, Equals, 33)
 	c.Check(s.config.DatabaseOpenAttempts, Equals, 33)
+	c.Check(s.config.PublishRenameConcurrency, Equals, 3)
+}
+
+func (s *ConfigSuite) TestLoadConfigKeepsTheDefaultRenameConcurrency(c *C) {
+	configname := filepath.Join(c.MkDir(), "aptly.json3")
+	c.Assert(os.WriteFile(configname, []byte(`{"rootDir": "/opt/aptly/"}`), 0644), IsNil)
+
+	s.config = Config
+	c.Assert(LoadConfig(configname, &s.config), IsNil)
+	c.Check(s.config.PublishRenameConcurrency, Equals, 4)
 }
 
 func (s *ConfigSuite) TestSaveConfig(c *C) {
@@ -114,6 +124,7 @@ func (s *ConfigSuite) TestSaveConfig(c *C) {
   "gpgKeys": null,
   "skipContentsPublishing": false,
   "skipBz2Publishing": false,
+  "publishRenameConcurrency": 0,
   "FileSystemPublishEndpoints": {
     "test": {
       "rootDir": "/opt/aptly-publish",
@@ -213,6 +224,7 @@ func (s *ConfigSuite) TestLoadYAMLConfig(c *C) {
   c.Check(s.config.GetRootDir(), Equals, "/opt/aptly/")
   c.Check(s.config.DownloadConcurrency, Equals, 40)
   c.Check(s.config.DatabaseOpenAttempts, Equals, 10)
+  c.Check(s.config.PublishRenameConcurrency, Equals, 2)
 }
 
 func (s *ConfigSuite) TestLoadYAMLErrorConfig(c *C) {
@@ -311,6 +323,7 @@ func (s *ConfigSuite) TestSaveYAML2Config(c *C) {
     "gpg_keys: []\n"+
     "skip_contents_publishing: false\n"+
     "skip_bz2_publishing: false\n"+
+    "publish_rename_concurrency: 0\n"+
     "filesystem_publish_endpoints: {}\n"+
     "jfrog_publish_endpoints: {}\n"+
     "s3_publish_endpoints: {}\n"+
@@ -334,7 +347,7 @@ func (s *ConfigSuite) TestLoadEmptyConfig(c *C) {
 	c.Assert(err.Error(), Equals, "invalid yaml (EOF) or json (EOF)")
 }
 
-const configFile = `{"rootDir": "/opt/aptly/", "downloadConcurrency": 33, "databaseOpenAttempts": 33}`
+const configFile = `{"rootDir": "/opt/aptly/", "downloadConcurrency": 33, "databaseOpenAttempts": 33, "publishRenameConcurrency": 3}`
 const configFileYAML = `root_dir: /opt/aptly/
 log_level: error
 log_format: json
@@ -372,6 +385,7 @@ gpg_disable_verify: true
 gpg_keys: []
 skip_contents_publishing: true
 skip_bz2_publishing: true
+publish_rename_concurrency: 2
 filesystem_publish_endpoints:
     test1:
         root_dir: /opt/srv/aptly_public