Merge pull request #483 from turt2live/travis/fix-redis-perf

Improve error handling around redis cache population on upload

CHANGELOG.md

@@ -7,7 +7,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ## [Unreleased]
 
-*Nothing yet.*
+### Fixed
+
+* Improved handling when encountering an error attempting to populate Redis during uploads.
 
 ## [1.3.2] - September 13, 2023
 

pipelines/_steps/upload/redis_async.go

@@ -3,6 +3,7 @@ package upload
 import (
 	"io"
 
+	"github.com/getsentry/sentry-go"
 	"github.com/turt2live/matrix-media-repo/common/rcontext"
 	"github.com/turt2live/matrix-media-repo/redislib"
 )
@@ -18,6 +19,7 @@ func PopulateCacheAsync(ctx rcontext.RequestContext, reader io.Reader, size int6
 		err = redislib.StoreMedia(ctx, sha256hash, reader, size)
 		if err != nil {
 			ctx.Log.Debug("Not populating cache due to error: ", err)
+			sentry.CaptureException(err)
 			return
 		}
 	}()

redislib/cache.go

@@ -39,19 +39,31 @@ func StoreMedia(ctx rcontext.RequestContext, hash string, content io.Reader, siz
 		return err
 	}
 
+	cleanup := func() {
+		if delErr := DeleteMedia(ctx, hash); delErr != nil {
+			ctx.Log.Warn("Error while attempting to clean up cache during another error: ", delErr)
+			sentry.CaptureException(delErr)
+		}
+	}
+
 	buf := make([]byte, appendBufferSize)
 	for {
 		read, err := content.Read(buf)
-		eof := errors.Is(err, io.EOF)
+		eof := false
+		if err != nil {
+			if errors.Is(err, io.EOF) {
+				eof = true
+			} else {
+				cleanup()
+				return err
+			}
+		}
 		if read > 0 {
 			if err = ring.ForEachShard(ctx.Context, func(ctx2 context.Context, client *redis.Client) error {
 				res := client.Append(ctx2, hash, string(buf[0:read]))
 				return res.Err()
 			}); err != nil {
-				if delErr := DeleteMedia(ctx, hash); delErr != nil {
-					ctx.Log.Warn("Error while attempting to clean up cache during another error: ", delErr)
-					sentry.CaptureException(delErr)
-				}
+				cleanup()
 				return err
 			}
 		}