Merge pull request #2 from impossiblecloud/fix-for-slow-uploads

Add file locking to avoid double-processing

Author: adidenko

.github/PULL_REQUEST_TEMPLATE.md

@@ -1,5 +1,3 @@
-## Description
-
 <!--
 Please provide a meaningful description of what this change will do, or is for. Bonus points for including links to related issues, other PRs, or technical references.
 

internal/fs/fs.go

@@ -13,6 +13,8 @@ import (
 	"github.com/fsnotify/fsnotify"
 )
 
+const lockFilePrefix = "/tmp/s3-file-uploader.lock"
+
 // Check if fs event is the one we care about
 func isValidFsEvent(event fsnotify.Event) bool {
 
@@ -66,7 +68,12 @@ func fsScan(comm *chan cfg.Message, config cfg.AppConfig) {
 
 	for _, e := range entries {
 		//config.Applog.Infof("Found file %q", e.Name())
-		*comm <- cfg.Message{File: filepath.Join(config.PathToWatch, e.Name())}
+		filename := filepath.Join(config.PathToWatch, e.Name())
+		if IsLocked(filename) {
+			config.Applog.Infof("Found file %q but it's already being processed (lock detected)", filename)
+		} else {
+			*comm <- cfg.Message{File: filename}
+		}
 	}
 }
 
@@ -190,3 +197,39 @@ func DeleteFile(config cfg.AppConfig, filename string) error {
 
 	return nil
 }
+
+func getLockFileName(filename string) string {
+	file := filepath.Base(filename)
+	return fmt.Sprintf("%s.%s", lockFilePrefix, file)
+}
+
+// IsLocked checks if the file is locked
+func IsLocked(filename string) bool {
+	lockFile := getLockFileName(filename)
+	if _, err := os.Stat(lockFile); err != nil {
+		return false
+	}
+	return true
+}
+
+// Lock locks the file for exclusive access for processing
+func Lock(filename string, id int) error {
+	if IsLocked(filename) {
+		return fmt.Errorf("file %s is already locked", filename)
+	}
+
+	lockFile := getLockFileName(filename)
+	f, err := os.Create(lockFile)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	f.WriteString(fmt.Sprintf("%d", id))
+	return nil
+}
+
+// UnLock unlocks the file
+func UnLock(filename string) error {
+	lockFile := getLockFileName(filename)
+	return os.Remove(lockFile)
+}

internal/s3/s3.go

@@ -3,8 +3,10 @@ package s3
 import (
 	"fmt"
 	"io"
+	"math/rand/v2"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/impossiblecloud/s3-file-uploader/internal/cfg"
 	"github.com/impossiblecloud/s3-file-uploader/internal/utils"
@@ -74,6 +76,10 @@ func FakeUploadFile(config cfg.AppConfig, filename string) (int64, error) {
 		return 0, err
 	}
 
+	// Simulate random upload time by sleeping for a random duration
+	sleepSec := rand.IntN(10) + 5
+	time.Sleep(time.Duration(sleepSec) * time.Second)
+
 	config.Applog.Infof("FAKE UPLOAD TO S3: %q file, size %s", realFile, utils.HumanizeBytes(fi.Size(), false))
 	return fi.Size(), nil
 }

main.go

@@ -247,6 +247,7 @@ func worker(wg *sync.WaitGroup, ctx context.Context, id int, config cfg.AppConfi
 			}
 
 			applog.Infof("Worker %d: processing file %q", id, msg.File)
+			fs.Lock(msg.File, id)
 
 			config.Metrics.FileSendCount.WithLabelValues().Inc()
 			err := sendFileS3(config, client, msg.File)
@@ -256,6 +257,7 @@ func worker(wg *sync.WaitGroup, ctx context.Context, id int, config cfg.AppConfi
 			} else {
 				config.Metrics.FileSendSuccess.WithLabelValues().Inc()
 			}
+			fs.UnLock(msg.File)
 		}
 	}
 }