fixup! s3store: Use context to stop producing parts

pkg/s3store/s3store.go

@@ -463,7 +463,10 @@ func (upload *s3Upload) uploadParts(ctx context.Context, offset int64, src io.Re
 	partProducer, fileChan := newS3PartProducer(src, store.MaxBufferedParts, store.TemporaryDirectory, store.diskWriteDurationMetric)
 
 	producerCtx, cancelProducer := context.WithCancel(ctx)
-	defer cancelProducer()
+	defer func() {
+		cancelProducer()
+		partProducer.closeUnreadFiles()
+	}()
 	go partProducer.produce(producerCtx, optimalPartSize)
 
 	var wg sync.WaitGroup

pkg/s3store/s3store_part_producer.go

@@ -45,6 +45,16 @@ func newS3PartProducer(source io.Reader, backlog int64, tmpDir string, diskWrite
 	return partProducer, fileChan
 }
 
+// closeUnreadFiles should always be called by the consumer to ensure that the channels
+// are properly closed and emptied.
+func (spp *s3PartProducer) closeUnreadFiles() {
+	// If we return while there are still files in the channel, then
+	// we may leak file descriptors. Let's ensure that those are cleaned up.
+	for fileChunk := range spp.files {
+		fileChunk.closeReader()
+	}
+}
+
 func (spp *s3PartProducer) produce(ctx context.Context, partSize int64) {
 outerloop:
 	for {
@@ -66,13 +76,7 @@ outerloop:
 		}
 	}
 
-	// First, close the channel.
 	close(spp.files)
-
-	// And then empty all entries to ensure that we close the file descriptors.
-	for fileChunk := range spp.files {
-		fileChunk.closeReader()
-	}
 }
 
 func (spp *s3PartProducer) nextPart(size int64) (fileChunk, bool, error) {