fix: properly detect created hls segments

Signed-off-by: Alexander Onnikov <Alexander.Onnikov@xored.com>

Dockerfile

@@ -11,7 +11,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM golang:1.24.1 AS builder
+FROM golang:1.24.4 AS builder
 ENV GO111MODULE=on
 ENV CGO_ENABLED=0
 ENV GOBIN=/bin

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/credentials v1.17.59
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.77.0
 	github.com/getsentry/sentry-go v0.31.1
-	github.com/golang-jwt/jwt/v5 v5.2.1
+	github.com/golang-jwt/jwt/v5 v5.2.2
 	github.com/google/uuid v1.6.0
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/pkg/errors v0.9.1

go.sum

@@ -45,8 +45,8 @@ github.com/getsentry/sentry-go v0.31.1 h1:ELVc0h7gwyhnXHDouXkhqTFSO5oslsRDk0++ey
 github.com/getsentry/sentry-go v0.31.1/go.mod h1:CYNcMMz73YigoHljQRG+qPF+eMq8gG72XcGN/p71BAY=
 github.com/go-errors/errors v1.4.2 h1:J6MZopCL4uSllY1OfXM374weqZFFItUbrImctkmUxIA=
 github.com/go-errors/errors v1.4.2/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og=
-github.com/golang-jwt/jwt/v5 v5.2.1 h1:OuVbFODueb089Lh128TAcimifWaLhJwVflnrgM17wHk=
-github.com/golang-jwt/jwt/v5 v5.2.1/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk=
+github.com/golang-jwt/jwt/v5 v5.2.2 h1:Rl4B7itRWVtYIHFrSNd7vhTiz9UpLdi6gZhZ3wEeDy8=
+github.com/golang-jwt/jwt/v5 v5.2.2/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk=
 github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
 github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=

internal/pkg/mediaconvert/command.go

@@ -19,7 +19,6 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"os"
 	"os/exec"
 	"path/filepath"
 	"strings"
@@ -30,12 +29,27 @@ import (
 	"go.uber.org/zap"
 )
 
+type LogLevel string
+
+const (
+	LogLevelQuiet   LogLevel = "quiet"
+	LogLevelPanic   LogLevel = "panic"
+	LogLevelFatal   LogLevel = "fatal"
+	LogLevelError   LogLevel = "error"
+	LogLevelWarning LogLevel = "warning"
+	LogLevelInfo    LogLevel = "info"
+	LogLevelVerbose LogLevel = "verbose"
+	LogLevelDebug   LogLevel = "debug"
+	LogLevelTrace   LogLevel = "trace"
+)
+
 // Options represents configuration for the ffmpeg command
 type Options struct {
 	Input         string
 	OutputDir     string
 	ScalingLevels []string
 	Level         string
+	LogLevel      LogLevel
 	Transcode     bool
 	Threads       int
 	UploadID      string
@@ -51,8 +65,6 @@ func newFfmpegCommand(ctx context.Context, in io.Reader, args []string) (*exec.C
 	logger.Debug("prepared command: ", zap.Strings("args", args))
 
 	var result = exec.CommandContext(ctx, "ffmpeg", args...)
-	result.Stderr = os.Stdout
-	result.Stdout = os.Stdout
 	result.Stdin = in
 
 	return result, nil
@@ -60,6 +72,8 @@ func newFfmpegCommand(ctx context.Context, in io.Reader, args []string) (*exec.C
 
 func buildCommonCommand(opts *Options) []string {
 	var result = []string{
+		"-y", // Overwrite output files without asking.
+		"-v", string(opts.LogLevel),
 		"-threads", fmt.Sprint(opts.Threads),
 		"-i", opts.Input,
 	}
@@ -96,7 +110,7 @@ func BuildRawVideoCommand(opts *Options) []string {
 			"-g", "60",
 			"-f", "hls",
 			"-hls_time", "5",
-			"-hls_flags", "split_by_time",
+			"-hls_flags", "split_by_time+temp_file",
 			"-hls_list_size", "0",
 			"-hls_segment_filename", filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_%s.ts", opts.UploadID, "%03d", opts.Level)),
 			filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_master.m3u8", opts.UploadID, opts.Level)))
@@ -107,7 +121,7 @@ func BuildRawVideoCommand(opts *Options) []string {
 		"-c:v", "copy", // Copy video stream
 		"-f", "hls",
 		"-hls_time", "5",
-		"-hls_flags", "split_by_time",
+		"-hls_flags", "split_by_time+temp_file",
 		"-hls_list_size", "0",
 		"-hls_segment_filename", filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_%s.ts", opts.UploadID, "%03d", opts.Level)),
 		filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_master.m3u8", opts.UploadID, opts.Level)))
@@ -142,7 +156,15 @@ func BuildScalingVideoCommand(opts *Options) []string {
 			"-g", "60",
 			"-f", "hls",
 			"-hls_time", "5",
-			"-hls_flags", "split_by_time",
+			// Use HLS flags
+			// - split_by_time
+			//     Allow segments to start on frames other than key frames.
+			//     This improves behavior on some players when the time between key frames is inconsistent,
+			//     but may make things worse on others, and can cause some oddities during seeking.
+			//     This flag should be used with the hls_time option.
+			// - temp_file
+			//     Write segment data to filename.tmp and rename to filename only once the segment is complete.
+			"-hls_flags", "split_by_time+temp_file",
 			"-hls_list_size", "0",
 			"-hls_segment_filename", filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_%s.ts", opts.UploadID, "%03d", level)),
 			filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_master.m3u8", opts.UploadID, level)))

internal/pkg/mediaconvert/command_test.go

@@ -31,7 +31,7 @@ func Test_BuildVideoCommand_Scaling(t *testing.T) {
 		ScalingLevels: []string{"720p", "480p"},
 	})
 
-	const expected = `-threads 4 -i pipe:0 -map 0:v -vf scale=-2:720 -c:a aac -c:v libx264 -preset veryfast -crf 23 -g 60 -f hls -hls_time 5 -hls_flags split_by_time -hls_list_size 0 -hls_segment_filename test/1/1_%03d_720p.ts test/1/1_720p_master.m3u8 -map 0:v -vf scale=-2:480 -c:a aac -c:v libx264 -preset veryfast -crf 23 -g 60 -f hls -hls_time 5 -hls_flags split_by_time -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8`
+	const expected = `-threads 4 -i pipe:0 -map 0:v -vf scale=-2:720 -c:a aac -c:v libx264 -preset veryfast -crf 23 -g 60 -f hls -hls_time 5 -hls_flags split_by_time+temp_file -hls_list_size 0 -hls_segment_filename test/1/1_%03d_720p.ts test/1/1_720p_master.m3u8 -map 0:v -vf scale=-2:480 -c:a aac -c:v libx264 -preset veryfast -crf 23 -g 60 -f hls -hls_time 5 -hls_flags split_by_time -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8`
 
 	require.Contains(t, expected, strings.Join(scaleCommand, " "))
 }
@@ -46,7 +46,7 @@ func Test_BuildVideoCommand_Scaling_NoRaw(t *testing.T) {
 		ScalingLevels: []string{"720p", "480p"},
 	})
 
-	const expected = `-threads 4 -i pipe:0 -map 0:v -vf scale=-2:480 -c:a aac -c:v libx264 -preset veryfast -crf 23 -g 60 -f hls -hls_time 5 -hls_flags split_by_time -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8`
+	const expected = `-threads 4 -i pipe:0 -map 0:v -vf scale=-2:480 -c:a aac -c:v libx264 -preset veryfast -crf 23 -g 60 -f hls -hls_time 5 -hls_flags split_by_time+temp_file -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8`
 
 	require.Contains(t, expected, strings.Join(scaleCommand, " "))
 }
@@ -61,7 +61,7 @@ func Test_BuildVideoCommand_Raw_NoTranscode(t *testing.T) {
 		Transcode: false,
 	})
 
-	const expected = `"-threads 4 -i pipe:0 -c:a copy -c:v copy -f hls -hls_time 5 -hls_flags split_by_time -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8`
+	const expected = `"-threads 4 -i pipe:0 -c:a copy -c:v copy -f hls -hls_time 5 -hls_flags split_by_time+temp_file -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8`
 
 	require.Contains(t, expected, strings.Join(rawCommand, " "))
 }
@@ -76,7 +76,7 @@ func Test_BuildVideoCommand_Raw_Transcode(t *testing.T) {
 		Transcode: true,
 	})
 
-	const expected = `-threads 4 -i pipe:0 -c:a aac -c:v libx264 -preset veryfast -crf 23 -g 60 -f hls -hls_time 5 -hls_flags split_by_time -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8`
+	const expected = `-threads 4 -i pipe:0 -c:a aac -c:v libx264 -preset veryfast -crf 23 -g 60 -f hls -hls_time 5 -hls_flags split_by_time+temp_file -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8`
 
 	require.Contains(t, expected, strings.Join(rawCommand, " "))
 }

internal/pkg/mediaconvert/scheduler.go

@@ -299,6 +299,8 @@ func IsSupportedMediaType(mediaType string) bool {
 		return true
 	case "video/webm":
 		return true
+	case "video/quicktime":
+		return true
 	default:
 		return false
 	}

internal/pkg/mediaconvert/transcoder.go

@@ -16,8 +16,10 @@
 package mediaconvert
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -42,6 +44,12 @@ type Transcoder struct {
 	logger *zap.Logger
 }
 
+type Command struct {
+	cmd       *exec.Cmd
+	stdoutBuf bytes.Buffer
+	stderrBuf bytes.Buffer
+}
+
 // NewTranscoder creates a new instance of task transcoder
 func NewTranscoder(ctx context.Context, cfg *config.Config) *Transcoder {
 	var p = &Transcoder{
@@ -54,7 +62,7 @@ func NewTranscoder(ctx context.Context, cfg *config.Config) *Transcoder {
 }
 
 // Transcode handles one transcoding task
-func (p *Transcoder) Transcode(ctx context.Context, task *Task) (TaskResult, error) {
+func (p *Transcoder) Transcode(ctx context.Context, task *Task) (*TaskResult, error) {
 	var logger = p.logger.With(zap.String("task-id", task.ID))
 
 	logger.Debug("start")
@@ -64,7 +72,7 @@ func (p *Transcoder) Transcode(ctx context.Context, task *Task) (TaskResult, err
 	var tokenString, err = token.NewToken(p.cfg.ServerSecret, task.Workspace, "stream", "datalake")
 	if err != nil {
 		logger.Error("can not create token", zap.Error(err))
-		return TaskResult{}, errors.Wrapf(err, "can not create token")
+		return nil, errors.Wrapf(err, "can not create token")
 	}
 
 	logger.Debug("phase 2: preparing fs")
@@ -73,7 +81,7 @@ func (p *Transcoder) Transcode(ctx context.Context, task *Task) (TaskResult, err
 	err = os.MkdirAll(destinationFolder, os.ModePerm)
 	if err != nil {
 		logger.Error("can not create temporary folder", zap.Error(err))
-		return TaskResult{}, errors.Wrapf(err, "can not create temporary folder")
+		return nil, errors.Wrapf(err, "can not create temporary folder")
 	}
 
 	defer func() {
@@ -87,38 +95,38 @@ func (p *Transcoder) Transcode(ctx context.Context, task *Task) (TaskResult, err
 	remoteStorage, err := storage.NewStorageByURL(ctx, p.cfg.Endpoint(), p.cfg.EndpointURL.Scheme, tokenString, task.Workspace)
 	if err != nil {
 		logger.Error("can not create storage by url", zap.Error(err), zap.String("url", p.cfg.EndpointURL.String()))
-		return TaskResult{}, errors.Wrapf(err, "can not create storage by url")
+		return nil, errors.Wrapf(err, "can not create storage by url")
 	}
 
 	stat, err := remoteStorage.StatFile(ctx, task.Source)
 	if err != nil {
 		logger.Error("can not stat file", zap.Error(err), zap.String("filepath", task.Source))
-		return TaskResult{}, errors.Wrapf(err, "can not stat file")
+		return nil, errors.Wrapf(err, "can not stat file")
 	}
 
 	if !IsSupportedMediaType(stat.Type) {
 		logger.Info("unsupported media type", zap.String("type", stat.Type))
-		return TaskResult{}, fmt.Errorf("unsupported media type: %s", stat.Type)
+		return nil, fmt.Errorf("unsupported media type: %s", stat.Type)
 	}
 
 	sourceFilePath := filepath.Join(destinationFolder, filename)
 	if err = remoteStorage.GetFile(ctx, task.Source, sourceFilePath); err != nil {
 		logger.Error("can not download source file", zap.Error(err), zap.String("filepath", task.Source))
 		// TODO: reschedule
-		return TaskResult{}, errors.Wrapf(err, "can not download source file")
+		return nil, errors.Wrapf(err, "can not download source file")
 	}
 
 	logger.Debug("phase 4: prepare to transcode")
 	probe, err := ffprobe.ProbeURL(ctx, sourceFilePath)
 	if err != nil {
 		logger.Error("can not get ffprobe", zap.Error(err), zap.String("filepath", sourceFilePath))
-		return TaskResult{}, errors.Wrapf(err, "can not get ffprobe")
+		return nil, errors.Wrapf(err, "can not get ffprobe")
 	}
 
 	videoStream := probe.FirstVideoStream()
 	if videoStream == nil {
 		logger.Error("no video stream found", zap.String("filepath", sourceFilePath))
-		return TaskResult{}, errors.Wrapf(err, "no video stream found")
+		return nil, errors.Wrapf(err, "no video stream found")
 	}
 
 	logger.Debug("video stream found", zap.String("codec", videoStream.CodecName), zap.Int("width", videoStream.Width), zap.Int("height", videoStream.Height))
@@ -136,6 +144,7 @@ func (p *Transcoder) Transcode(ctx context.Context, task *Task) (TaskResult, err
 		Input:         sourceFilePath,
 		OutputDir:     p.cfg.OutputDir,
 		Level:         level,
+		LogLevel:      LogLevel(p.cfg.LogLevel),
 		Transcode:     !IsHLSSupportedVideoCodec(codec),
 		ScalingLevels: append(sublevels, level),
 		UploadID:      task.ID,
@@ -157,7 +166,7 @@ func (p *Transcoder) Transcode(ctx context.Context, task *Task) (TaskResult, err
 	err = manifest.GenerateHLSPlaylist(opts.ScalingLevels, p.cfg.OutputDir, opts.UploadID)
 	if err != nil {
 		logger.Error("can not generate hls playlist", zap.String("out", p.cfg.OutputDir), zap.String("uploadID", opts.UploadID))
-		return TaskResult{}, errors.Wrapf(err, "can not generate hls playlist")
+		return nil, errors.Wrapf(err, "can not generate hls playlist")
 	}
 
 	go uploader.Start()
@@ -169,29 +178,42 @@ func (p *Transcoder) Transcode(ctx context.Context, task *Task) (TaskResult, err
 		BuildRawVideoCommand(&opts),
 		BuildScalingVideoCommand(&opts),
 	}
-	var cmds []*exec.Cmd
+	var cmds []Command
 
 	for _, args := range argsSlice {
 		cmd, cmdErr := newFfmpegCommand(ctx, nil, args)
 		if cmdErr != nil {
 			logger.Error("can not create a new command", zap.Error(cmdErr), zap.Strings("args", args))
 			go uploader.Cancel()
-			return TaskResult{}, errors.Wrapf(err, "can not create a new command")
+			return nil, errors.Wrapf(err, "can not create a new command")
 		}
-		cmds = append(cmds, cmd)
+
+		var command = Command{
+			cmd:       cmd,
+			stdoutBuf: bytes.Buffer{},
+			stderrBuf: bytes.Buffer{},
+		}
+
+		cmd.Stdout = io.MultiWriter(os.Stdout, &command.stdoutBuf)
+		cmd.Stderr = io.MultiWriter(os.Stderr, &command.stderrBuf)
+
+		cmds = append(cmds, command)
 		if err = cmd.Start(); err != nil {
 			logger.Error("can not start a command", zap.Error(err), zap.Strings("args", args))
 			go uploader.Cancel()
-			return TaskResult{}, errors.Wrapf(err, "can not start a command")
+			return nil, errors.Wrapf(err, "can not start a command")
 		}
 	}
 
 	logger.Debug("phase 7: wait for the result")
+
 	for _, cmd := range cmds {
-		if err = cmd.Wait(); err != nil {
+		if err = cmd.cmd.Wait(); err != nil {
 			logger.Error("can not wait for command end ", zap.Error(err))
+			os.Stdout.Write(cmd.stdoutBuf.Bytes())
+			os.Stderr.Write(cmd.stderrBuf.Bytes())
 			go uploader.Cancel()
-			return TaskResult{}, errors.Wrapf(err, "can not wait for command end")
+			return nil, errors.Wrapf(err, "can not wait for command end")
 		}
 	}
 
@@ -226,5 +248,5 @@ func (p *Transcoder) Transcode(ctx context.Context, task *Task) (TaskResult, err
 		}
 	}
 
-	return result, nil
+	return &result, nil
 }

internal/pkg/storage/datalake.go

@@ -21,8 +21,10 @@ import (
 	"io"
 	"mime/multipart"
 	"net/textproto"
+	"net/url"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"time"
 
@@ -170,6 +172,11 @@ func (d *DatalakeStorage) DeleteFile(ctx context.Context, fileName string) error
 		return errors.Wrapf(err, "delete failed")
 	}
 
+	if err := okResponse(res); err != nil {
+		logRequestError(logger, err, "bad status code", res)
+		return err
+	}
+
 	logger.Debug("deleted")
 
 	return nil
@@ -205,14 +212,11 @@ func (d *DatalakeStorage) PatchMeta(ctx context.Context, filename string, md *Me
 		return err
 	}
 
-	if resp.StatusCode() != fasthttp.StatusOK {
-		var err = fmt.Errorf("unexpected status code: %d", resp.StatusCode())
-		logger.Debug("bad status code", zap.Error(err))
+	if err := okResponse(resp); err != nil {
+		logRequestError(logger, err, "bad status code", resp)
 		return err
 	}
 
-	fmt.Println(string(resp.Body()))
-
 	return nil
 }
 
@@ -237,9 +241,8 @@ func (d *DatalakeStorage) GetMeta(ctx context.Context, filename string) (*Metada
 		return nil, err
 	}
 
-	if resp.StatusCode() != fasthttp.StatusOK {
-		var err = fmt.Errorf("unexpected status code: %d", resp.StatusCode())
-		logger.Debug("bad status code", zap.Error(err))
+	if err := okResponse(resp); err != nil {
+		logRequestError(logger, err, "bad status code", resp)
 		return nil, err
 	}
 
@@ -270,10 +273,8 @@ func (d *DatalakeStorage) GetFile(ctx context.Context, filename, destination str
 		return err
 	}
 
-	// Check the response status code
-	if resp.StatusCode() != fasthttp.StatusOK {
-		var err = fmt.Errorf("unexpected status code: %d", resp.StatusCode())
-		logger.Debug("bad status code", zap.Error(err))
+	if err := okResponse(resp); err != nil {
+		logRequestError(logger, err, "bad status code", resp)
 		return err
 	}
 
@@ -291,7 +292,13 @@ func (d *DatalakeStorage) GetFile(ctx context.Context, filename, destination str
 		return err
 	}
 
-	logger.Debug("file downloaded successfully")
+	stat, err := os.Stat(destination)
+	if err != nil {
+		logger.Error("can't stat the file", zap.Error(err))
+		return err
+	}
+
+	logger.Info("file downloaded successfully", zap.Int64("size", stat.Size()))
 	return nil
 }
 
@@ -316,9 +323,7 @@ func (d *DatalakeStorage) StatFile(ctx context.Context, filename string) (*BlobI
 		return nil, err
 	}
 
-	// Check the response status code
-	if resp.StatusCode() != fasthttp.StatusOK {
-		var err = fmt.Errorf("unexpected status code: %d", resp.StatusCode())
+	if err := okResponse(resp); err != nil {
 		logRequestError(logger, err, "bad status code", resp)
 		return nil, err
 	}
@@ -346,7 +351,7 @@ func (d *DatalakeStorage) SetParent(ctx context.Context, filename, parent string
 	req.SetRequestURI(d.baseURL + "/blob/" + d.workspace + "/" + objectKey + "/parent")
 	req.Header.SetMethod(fasthttp.MethodPatch)
 	req.Header.Add("Authorization", "Bearer "+d.token)
-	req.Header.Add("Content-Type", "application/json")
+	req.Header.SetContentType("application/json")
 
 	body := map[string]any{
 		"parent": parentKey,
@@ -365,15 +370,20 @@ func (d *DatalakeStorage) SetParent(ctx context.Context, filename, parent string
 		return err
 	}
 
-	// Check the response status code
-	var statusOK = resp.StatusCode() >= 200 && resp.StatusCode() < 300
-	if !statusOK {
-		var err = fmt.Errorf("unexpected status code: %d", resp.StatusCode())
-		logger.Debug("bad status code", zap.Error(err), zap.Int("status", resp.StatusCode()), zap.String("response", resp.String()))
+	if err := okResponse(resp); err != nil {
+		logRequestError(logger, err, "bad status code", resp)
 		return err
 	}
 
-	logger.Debug("finished")
+	return nil
+}
+
+func okResponse(res *fasthttp.Response) error {
+	var statusOK = res.StatusCode() >= 200 && res.StatusCode() < 300
+
+	if !statusOK {
+		return fmt.Errorf("unexpected status code: %d", res.StatusCode())
+	}
 
 	return nil
 }

internal/pkg/uploader/uploader.go

@@ -35,6 +35,8 @@ import (
 // See at https://man7.org/linux/man-pages/man7/inotify.7.html
 const inotifyCloseWrite uint32 = 0x8 // IN_CLOSE_WRITE
 const inotifyMovedTo uint32 = 0x80   // IN_MOVED_TO
+const inotifyDelete uint32 = 0x200   // IN_DELETE
+const inotifyMovedFrom uint32 = 0x40 // IN_MOVED_FROM
 
 // Uploader represents file uploader
 type Uploader interface {
@@ -250,9 +252,12 @@ func (u *uploaderImpl) uploadAndDelete(f string) {
 	}
 
 	// Check if the file exists
-	_, err := os.Stat(f)
-	if err != nil {
-		logger.Debug("file does not exist")
+	if _, err := os.Stat(f); err != nil {
+		if os.IsNotExist(err) {
+			logger.Debug("file does not exist", zap.Error(err))
+		} else {
+			logger.Error("failed to stat file", zap.Error(err))
+		}
 		return
 	}
 
@@ -300,6 +305,7 @@ func (u *uploaderImpl) uploadAndDelete(f string) {
 	}
 }
 
+// startWatch watches for changes in the directory and uploads created files
 func (u *uploaderImpl) startWatch(ready chan<- struct{}) {
 	defer close(u.watcherDoneCh)
 
@@ -317,7 +323,7 @@ func (u *uploaderImpl) startWatch(ready chan<- struct{}) {
 		}
 	}()
 
-	if err := watcher.AddWatch(u.options.Dir, inotifyCloseWrite|inotifyMovedTo); err != nil {
+	if err := watcher.AddWatch(u.options.Dir, inotifyCloseWrite|inotifyMovedTo|inotifyDelete|inotifyMovedFrom); err != nil {
 		logger.Error("can not start watching", zap.Error(err))
 		close(ready)
 		return
@@ -348,8 +354,19 @@ func (u *uploaderImpl) startWatch(ready chan<- struct{}) {
 			if strings.HasSuffix(event.Name, ".tmp") {
 				continue
 			}
+			if event.Mask&(inotifyDelete|inotifyMovedFrom) != 0 {
+				logger.Debug("file deleted or moved away", zap.String("event", event.Name), zap.Uint32("mask", event.Mask))
+				continue
+			}
+
 			logger.Debug("received an event", zap.String("event", event.Name), zap.Uint32("mask", event.Mask))
 
+			if _, err := os.Stat(event.Name); os.IsNotExist(err) {
+				logger.Warn("file does not exist", zap.String("file", event.Name))
+				// wait a bit for file operations to complete
+				time.Sleep(100 * time.Millisecond)
+			}
+
 			u.filesCh <- event.Name
 		case err, ok := <-watcher.Error:
 			if !ok {