Transfer - The runProducerConsumers method might terminate prematurely (

#1086)

artifactory/commands/transferfiles/manager.go

@@ -238,8 +238,13 @@ func runProducerConsumers(pcWrapper *producerConsumerWrapper) (executionErr erro
 	// Run() is a blocking method, so once all chunk builders are idle, the tasks queue closes and Run() stops running.
 	pcWrapper.chunkBuilderProducerConsumer.Run()
 	if pcWrapper.chunkUploaderProducerConsumer.IsStarted() {
+		// There might be a moment when the chunk uploader has no upload tasks.
+		// This circumstance might lead to setting the finish notification before completing all file uploads.
+		// To address this, we reset the finish notification to ensure no remaining upload tasks after the next finish notification.
+		pcWrapper.chunkUploaderProducerConsumer.ResetFinishNotificationIfActive()
 		// Wait till notified that the uploader finished its tasks, and it will not receive new tasks from the builder.
 		<-pcWrapper.chunkUploaderProducerConsumer.GetFinishedNotification()
+		log.Debug("Chunk uploaded producer consumer has completed all tasks. All files relevant to this phase have all been uploaded.")
 	}
 	// Close the tasks queue with Done().
 	pcWrapper.chunkUploaderProducerConsumer.Done()

artifactory/commands/transferfiles/manager_test.go

@@ -0,0 +1,37 @@
+package transferfiles
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRunProducerConsumers(t *testing.T) {
+	// Create the producer-consumers
+	producerConsumerWrapper := newProducerConsumerWrapper()
+
+	// Add 10 tasks for the chunkBuilderProducerConsumer. Each task provides a task to the chunkUploaderProducerConsumer.
+	for i := 0; i < 10; i++ {
+		_, err := producerConsumerWrapper.chunkBuilderProducerConsumer.AddTask(func(int) error {
+			time.Sleep(time.Millisecond * 100)
+			_, err := producerConsumerWrapper.chunkUploaderProducerConsumer.AddTask(
+				func(int) error {
+					time.Sleep(time.Millisecond)
+					return nil
+				},
+			)
+			assert.NoError(t, err)
+			return nil
+		})
+		assert.NoError(t, err)
+	}
+
+	// Run the producer-consumers
+	err := runProducerConsumers(&producerConsumerWrapper)
+	assert.NoError(t, err)
+
+	// Assert no active treads left in the producer-consumers
+	assert.Zero(t, producerConsumerWrapper.chunkBuilderProducerConsumer.ActiveThreads())
+	assert.Zero(t, producerConsumerWrapper.chunkUploaderProducerConsumer.ActiveThreads())
+}

go.mod

@@ -103,4 +103,4 @@ require (
 
 // replace github.com/jfrog/build-info-go => github.com/jfrog/build-info-go v1.8.9-0.20231220102935-c8776c613ad8
 
-// replace github.com/jfrog/gofrog => github.com/jfrog/gofrog v1.3.2-0.20231130091721-6d742be8bc7a
+replace github.com/jfrog/gofrog => github.com/jfrog/gofrog v1.3.3-0.20231223133729-ef57bd08cedc

go.sum

@@ -87,8 +87,8 @@ github.com/jedib0t/go-pretty/v6 v6.4.0 h1:YlI/2zYDrweA4MThiYMKtGRfT+2qZOO65ulej8
 github.com/jedib0t/go-pretty/v6 v6.4.0/go.mod h1:MgmISkTWDSFu0xOqiZ0mKNntMQ2mDgOcwOkwBEkMDJI=
 github.com/jfrog/build-info-go v1.9.18 h1:0RKeZtNWZjONX5j94aIPQSKMi1whP2evmGQymYF+5mA=
 github.com/jfrog/build-info-go v1.9.18/go.mod h1:/5VZXH2Ud0IK3cOFwPykjwPOaEcHhzzbjnRiou+YKpM=
-github.com/jfrog/gofrog v1.3.2 h1:TktKP+PdZdxjkYZxWWIq4DkTGSYtr9Slsy+egZpEhUY=
-github.com/jfrog/gofrog v1.3.2/go.mod h1:AQo5Fq0G9nDEF6icH7MYQK0iohR4HuEAXl8jaxRuT6Q=
+github.com/jfrog/gofrog v1.3.3-0.20231223133729-ef57bd08cedc h1:aKVJSydPVBhJkEncWM9Vc66lM2slRDG/4ILXERKewJ0=
+github.com/jfrog/gofrog v1.3.3-0.20231223133729-ef57bd08cedc/go.mod h1:AQo5Fq0G9nDEF6icH7MYQK0iohR4HuEAXl8jaxRuT6Q=
 github.com/jfrog/jfrog-apps-config v1.0.1 h1:mtv6k7g8A8BVhlHGlSveapqf4mJfonwvXYLipdsOFMY=
 github.com/jfrog/jfrog-apps-config v1.0.1/go.mod h1:8AIIr1oY9JuH5dylz2S6f8Ym2MaadPLR6noCBO4C22w=
 github.com/jfrog/jfrog-client-go v1.35.3 h1:Kf4mErh1tlbHzKz3941+d9vpEsPM2clgdOaYOKfNQGI=