fix(bing-ads): close CSV file handles in bulk upload zip flow

router/batchrouter/asyncdestinationmanager/bing-ads/audience/types.go

@@ -51,6 +51,7 @@ type DestinationConfig struct {
 type ActionFileInfo struct {
 	Action           string
 	CSVWriter        *csv.Writer
+	CSVFile          *os.File
 	CSVFilePath      string
 	ZipFilePath      string
 	SuccessfulJobIDs []int64

router/batchrouter/asyncdestinationmanager/bing-ads/audience/util.go

@@ -50,20 +50,27 @@ func createActionFile(audienceId, actionType string) (*ActionFileInfo, error) {
 	}
 	csvWriter, err := CreateActionFileTemplate(csvFile, audienceId, actionType)
 	if err != nil {
+		_ = csvFile.Close()
+		_ = os.Remove(csvFilePath)
 		return nil, err
 	}
 	return &ActionFileInfo{
 		Action:      actionType,
 		ZipFilePath: zipFilePath,
 		CSVFilePath: csvFilePath,
+		CSVFile:     csvFile,
 		CSVWriter:   csvWriter,
 	}, nil
 }
 
 func convertCsvToZip(actionFile *ActionFileInfo) error {
+	if actionFile.CSVFile != nil {
+		_ = actionFile.CSVFile.Close()
+		actionFile.CSVFile = nil
+	}
 	if actionFile.EventCount == 0 {
-		os.Remove(actionFile.CSVFilePath)
-		os.Remove(actionFile.ZipFilePath)
+		_ = os.Remove(actionFile.CSVFilePath)
+		_ = os.Remove(actionFile.ZipFilePath)
 		return nil
 	}
 	zipFile, err := os.Create(actionFile.ZipFilePath)
@@ -82,6 +89,8 @@ func convertCsvToZip(actionFile *ActionFileInfo) error {
 	if err != nil {
 		return err
 	}
+	defer csvFile.Close()
+
 	if _, err := csvFile.Seek(0, 0); err != nil {
 		return err
 	}
@@ -143,9 +152,18 @@ func (b *BingAdsBulkUploader) createZipFile(filePath, audienceId string) ([]*Act
 	defer textFile.Close()
 
 	actionFiles := map[string]*ActionFileInfo{}
+	closeActionFiles := func() {
+		for _, af := range actionFiles {
+			if af != nil && af.CSVFile != nil {
+				_ = af.CSVFile.Close()
+				af.CSVFile = nil
+			}
+		}
+	}
 	for _, actionType := range actionTypes {
 		actionFiles[actionType], err = createActionFile(audienceId, actionType)
 		if err != nil {
+			closeActionFiles()
 			return nil, err
 		}
 	}
@@ -155,6 +173,7 @@ func (b *BingAdsBulkUploader) createZipFile(filePath, audienceId string) ([]*Act
 		line := scanner.Text()
 		var data Data
 		if err := jsonrs.Unmarshal([]byte(line), &data); err != nil {
+			closeActionFiles()
 			return nil, err
 		}
 
@@ -167,12 +186,14 @@ func (b *BingAdsBulkUploader) createZipFile(filePath, audienceId string) ([]*Act
 		actionFile := actionFiles[data.Message.Action]
 		err := b.populateZipFile(actionFile, audienceId, line, data)
 		if err != nil {
+			closeActionFiles()
 			return nil, err
 		}
 
 	}
 	scannerErr := scanner.Err()
 	if scannerErr != nil {
+		closeActionFiles()
 		return nil, scannerErr
 	}
 	actionFilesList := []*ActionFileInfo{}

router/batchrouter/asyncdestinationmanager/bing-ads/audience/util_close_test.go

@@ -0,0 +1,60 @@
+package audience
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/rudderlabs/rudder-server/utils/misc"
+)
+
+func ensureAsyncDestLogsDir(t *testing.T) {
+	t.Helper()
+	tmpDirPath, err := misc.GetTmpDir()
+	require.NoError(t, err)
+	require.NoError(t, os.MkdirAll(filepath.Join(tmpDirPath, misc.RudderAsyncDestinationLogs), 0o755))
+}
+
+func TestConvertCsvToZipClosesHandles(t *testing.T) {
+	t.Parallel()
+	ensureAsyncDestLogsDir(t)
+
+	t.Run("empty event count closes write handle and removes csv", func(t *testing.T) {
+		t.Parallel()
+
+		actionFile, err := createActionFile("aud-1", "Add")
+		require.NoError(t, err)
+		require.NotNil(t, actionFile.CSVFile)
+
+		writeHandle := actionFile.CSVFile
+		require.NoError(t, convertCsvToZip(actionFile))
+		require.Nil(t, actionFile.CSVFile)
+
+		_, writeErr := writeHandle.Write([]byte("x"))
+		require.Error(t, writeErr)
+		_, statErr := os.Stat(actionFile.CSVFilePath)
+		require.True(t, os.IsNotExist(statErr))
+	})
+
+	t.Run("non-empty event count closes write handle and creates zip", func(t *testing.T) {
+		t.Parallel()
+
+		actionFile, err := createActionFile("aud-1", "Add")
+		require.NoError(t, err)
+		actionFile.EventCount = 1
+		actionFile.CSVWriter.Flush()
+
+		writeHandle := actionFile.CSVFile
+		require.NoError(t, convertCsvToZip(actionFile))
+		require.Nil(t, actionFile.CSVFile)
+
+		_, writeErr := writeHandle.Write([]byte("x"))
+		require.Error(t, writeErr)
+
+		_, err = os.Stat(actionFile.ZipFilePath)
+		require.NoError(t, err)
+		require.NoError(t, os.Remove(actionFile.ZipFilePath))
+	})
+}

router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/types.go

@@ -3,6 +3,7 @@ package offline_conversions
 import (
 	"encoding/csv"
 	"encoding/json"
+	"os"
 
 	"github.com/rudderlabs/bing-ads-go-sdk/bingads"
 	"github.com/rudderlabs/rudder-go-kit/logger"
@@ -42,6 +43,7 @@ type DestinationConfig struct {
 type ActionFileInfo struct {
 	Action           string
 	CSVWriter        *csv.Writer
+	CSVFile          *os.File
 	CSVFilePath      string
 	ZipFilePath      string
 	SuccessfulJobIDs []int64

router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/util.go

@@ -73,20 +73,27 @@ func createActionFile(actionType string) (*ActionFileInfo, error) {
 	}
 	csvWriter, err := CreateActionFileTemplate(csvFile, actionType)
 	if err != nil {
+		_ = csvFile.Close()
+		_ = os.Remove(csvFilePath)
 		return nil, err
 	}
 	return &ActionFileInfo{
 		Action:      actionType,
 		ZipFilePath: zipFilePath,
 		CSVFilePath: csvFilePath,
+		CSVFile:     csvFile,
 		CSVWriter:   csvWriter,
 	}, nil
 }
 
 func convertCsvToZip(actionFile *ActionFileInfo) error {
+	if actionFile.CSVFile != nil {
+		_ = actionFile.CSVFile.Close()
+		actionFile.CSVFile = nil
+	}
 	if actionFile.EventCount == 0 {
-		os.Remove(actionFile.CSVFilePath)
-		os.Remove(actionFile.ZipFilePath)
+		_ = os.Remove(actionFile.CSVFilePath)
+		_ = os.Remove(actionFile.ZipFilePath)
 		return nil
 	}
 	zipFile, err := os.Create(actionFile.ZipFilePath)
@@ -105,6 +112,8 @@ func convertCsvToZip(actionFile *ActionFileInfo) error {
 	if err != nil {
 		return err
 	}
+	defer csvFile.Close()
+
 	if _, err := csvFile.Seek(0, 0); err != nil {
 		return err
 	}
@@ -176,9 +185,18 @@ func (b *BingAdsBulkUploader) createZipFile(filePath string) ([]*ActionFileInfo,
 	}
 	defer textFile.Close()
 	actionFiles := map[string]*ActionFileInfo{}
+	closeActionFiles := func() {
+		for _, af := range actionFiles {
+			if af != nil && af.CSVFile != nil {
+				_ = af.CSVFile.Close()
+				af.CSVFile = nil
+			}
+		}
+	}
 	for _, actionType := range actionTypes {
 		actionFiles[actionType], err = createActionFile(actionType)
 		if err != nil {
+			closeActionFiles()
 			return nil, err
 		}
 	}
@@ -188,17 +206,20 @@ func (b *BingAdsBulkUploader) createZipFile(filePath string) ([]*ActionFileInfo,
 		line := scanner.Text()
 		var data Data
 		if err := jsonrs.Unmarshal([]byte(line), &data); err != nil {
+			closeActionFiles()
 			return nil, err
 		}
 		actionFile := actionFiles[data.Message.Action]
 		err := b.populateZipFile(actionFile, line, data)
 		if err != nil {
+			closeActionFiles()
 			return nil, err
 		}
 
 	}
 	scannerErr := scanner.Err()
 	if scannerErr != nil {
+		closeActionFiles()
 		return nil, scannerErr
 	}
 	actionFilesList := []*ActionFileInfo{}

router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/util_close_test.go

@@ -0,0 +1,60 @@
+package offline_conversions
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/rudderlabs/rudder-server/utils/misc"
+)
+
+func ensureAsyncDestLogsDir(t *testing.T) {
+	t.Helper()
+	tmpDirPath, err := misc.GetTmpDir()
+	require.NoError(t, err)
+	require.NoError(t, os.MkdirAll(filepath.Join(tmpDirPath, misc.RudderAsyncDestinationLogs), 0o755))
+}
+
+func TestConvertCsvToZipClosesHandles(t *testing.T) {
+	t.Parallel()
+	ensureAsyncDestLogsDir(t)
+
+	t.Run("empty event count closes write handle and removes csv", func(t *testing.T) {
+		t.Parallel()
+
+		actionFile, err := createActionFile("insert")
+		require.NoError(t, err)
+		require.NotNil(t, actionFile.CSVFile)
+
+		writeHandle := actionFile.CSVFile
+		require.NoError(t, convertCsvToZip(actionFile))
+		require.Nil(t, actionFile.CSVFile)
+
+		_, writeErr := writeHandle.Write([]byte("x"))
+		require.Error(t, writeErr)
+		_, statErr := os.Stat(actionFile.CSVFilePath)
+		require.True(t, os.IsNotExist(statErr))
+	})
+
+	t.Run("non-empty event count closes write handle and creates zip", func(t *testing.T) {
+		t.Parallel()
+
+		actionFile, err := createActionFile("insert")
+		require.NoError(t, err)
+		actionFile.EventCount = 1
+		actionFile.CSVWriter.Flush()
+
+		writeHandle := actionFile.CSVFile
+		require.NoError(t, convertCsvToZip(actionFile))
+		require.Nil(t, actionFile.CSVFile)
+
+		_, writeErr := writeHandle.Write([]byte("x"))
+		require.Error(t, writeErr)
+
+		_, err = os.Stat(actionFile.ZipFilePath)
+		require.NoError(t, err)
+		require.NoError(t, os.Remove(actionFile.ZipFilePath))
+	})
+}