fix: redis lock improvments

Author: aldor007

Dockerfile.build

@@ -1,25 +1,21 @@
 FROM ghcr.io/aldor007/mort-base 
 
-ENV GOLANG_VERSION 1.16.6
-ARG TARGETARCH amd64
+ENV GOLANG_VERSION 1.19.2
+ENV TARGETARCH amd64
 ARG TAG 'dev'
 ARG COMMIT "master"
 ARG DATE "now"
 
-ENV GOLANG_DOWNLOAD_URL https://golang.org/dl/go$GOLANG_VERSION.linux-$TARGETARCH.tar.gz
-
-RUN curl -fsSL --insecure "$GOLANG_DOWNLOAD_URL" -o golang.tar.gz \
-  && tar -C /usr/local -xzf golang.tar.gz \
-  && rm golang.tar.gz
 
 ENV WORKDIR /workspace
 ENV PATH /usr/local/go/bin:$PATH
-
+RUN rm -rf /usr/local/go && curl -fsSL --insecure "https://go.dev/dl/go$GOLANG_VERSION.linux-$TARGETARCH.tar.gz" -o golang.tar.gz \
+  && tar -C /usr/local -xzf golang.tar.gz \
+  && rm golang.tar.gz
 
 WORKDIR $WORKDIR
 COPY go.mod  ./
 COPY go.sum ./
-RUN go mod  download
 
 COPY cmd/  $WORKDIR/cmd
 COPY .godir ${WORKDIR}/.godir
@@ -28,3 +24,5 @@ COPY etc/ ${WORKDIR}/etc
 COPY pkg/ ${WORKDIR}/pkg
 COPY scripts/ ${WORKDIR}/scripts
 COPY Makefile ${WORKDIR}/Makefile
+
+RUN go build -o /go/mort ./cmd/mort/mort.go

Dockerfile.test

@@ -1,6 +1,8 @@
 package lock
 
 import (
+	"github.com/aldor007/mort/pkg/monitoring"
+	"go.uber.org/zap"
 	"sync"
 	"time"
 
@@ -34,6 +36,7 @@ type rediser interface {
 }
 
 type internalLockRedis struct {
+	lockData
 	lock   *redislock.Lock
 	pubsub *goRedis.PubSub
 }
@@ -86,11 +89,15 @@ func (m *RedisLock) NotifyAndRelease(ctx context.Context, key string, originalRe
 	defer m.lock.Unlock()
 	lock, ok := m.locks[key]
 	if ok {
-		lock.lock.Release(ctx)
+		err := lock.lock.Release(ctx)
+		if err != nil {
+			monitoring.Log().Error("redis release error", zap.String("key", key), zap.Error(err))
+		}
 		delete(m.locks, key)
 		if lock.pubsub != nil {
-			m.redisClient.Publish(ctx, key, 1)
 			lock.pubsub.Close()
+		} else {
+			m.redisClient.Publish(ctx, key, 1)
 		}
 	}
 
@@ -106,41 +113,45 @@ func (m *RedisLock) Lock(ctx context.Context, key string) (result LockResult, ok
 		lock.lock.Refresh(ctx, time.Second*time.Duration(m.LockTimeout/2), nil)
 	} else {
 		lock, err := m.client.Obtain(ctx, key, time.Duration(m.LockTimeout)*time.Second, nil)
-		pubsub := m.redisClient.Subscribe(ctx, key)
-		lockData := internalLockRedis{lock: lock, pubsub: pubsub}
+		lockData := internalLockRedis{lock: lock, pubsub: nil}
 
+		ok = false
 		if err == redislock.ErrNotObtained {
 			result, ok = m.memoryLock.forceLockAndAddWatch(ctx, key)
-			ok = false
+			pubsub := m.redisClient.Subscribe(ctx, key)
+			lockData.pubsub = pubsub
 			// Go channel which receives messages.
 			ch := pubsub.Channel()
 
 			go func() {
 				for {
 					select {
 					case <-ch:
-						m.NotifyAndRelease(ctx, key, nil)
+						m.memoryLock.NotifyAndRelease(ctx, key, nil)
 						return
 					case <-result.Cancel:
 						m.memoryLock.Release(ctx, key)
-						pubsub.Close()
+						err := pubsub.Close()
+						if err != nil {
+							monitoring.Log().Error("Redis lock pubsub err", zap.String("key", key), zap.Error(err))
+						}
 						return
 					case <-ctx.Done():
 						m.memoryLock.Release(ctx, key)
-						pubsub.Close()
+						err := pubsub.Close()
+						if err != nil {
+							monitoring.Log().Error("Redis lock pubsub err", zap.String("key", key), zap.Error(err))
+						}
 						return
 					}
 				}
 			}()
 
-			return
-
 		} else if err != nil {
 			result.Error = err
 			return
 		}
 		m.locks[key] = lockData
-
 	}
 	return m.memoryLock.Lock(ctx, key)
 }

pkg/lock/redis.go

@@ -142,3 +142,38 @@ func TestRedisLock_NotifyAndReleaseTwoInstancesOfLock(t *testing.T) {
 
 	}
 }
+func TestRedisLock_Cancel(t *testing.T) {
+	s := miniredis.RunT(t)
+
+	l := NewRedisLock([]string{s.Addr()}, nil)
+	key := "kluczi22"
+	ctx := context.Background()
+	c, acquired := l.Lock(ctx, key)
+
+	assert.True(t, acquired, "Should acquire lock")
+	assert.Nil(t, c.ResponseChan, "shouldn't return channel")
+
+	l2 := NewRedisLock([]string{s.Addr()}, nil)
+	result, lock := l2.Lock(ctx, key)
+
+	assert.False(t, lock, "Shouldn't acquire lock")
+	assert.NotNil(t, result.ResponseChan, "should return channel")
+
+	go func() {
+		c.Cancel <- true
+	}()
+
+	buf := make([]byte, 1000)
+	go l.NotifyAndRelease(ctx, key, response.NewBuf(200, buf))
+
+	timer := time.NewTimer(time.Second * 2)
+	select {
+	case <-timer.C:
+		t.Fatalf("timeout waiting for lock")
+		return
+	case res, ok := <-result.ResponseChan:
+		assert.False(t, ok, "channel shouldn't be closed")
+		assert.Nil(t, res, "should be nil")
+
+	}
+}

pkg/lock/redis_test.go

@@ -62,7 +62,6 @@ type requestMessage struct {
 	responseChan chan *response.Response
 	obj          *object.FileObject
 	request      *http.Request
-	cancel       chan struct{}
 }
 
 // Process handle incoming request and create response
@@ -72,7 +71,7 @@ func (r *RequestProcessor) Process(req *http.Request, obj *object.FileObject) *r
 	var timeout context.CancelFunc
 
 	// if object has transfroms we can assume that it is image generated by mort
-	// othervise it can be big zip file
+	// otherwise it can be big zip file
 	if obj.HasTransform() {
 		ctx, timeout = context.WithTimeout(pCtx, r.processTimeout)
 		defer timeout()
@@ -85,13 +84,11 @@ func (r *RequestProcessor) Process(req *http.Request, obj *object.FileObject) *r
 	msg.request = req
 	msg.obj = obj
 	msg.responseChan = make(chan *response.Response)
-	msg.cancel = make(chan struct{}, 1)
 
 	go r.processChan(ctx, msg)
 
 	select {
 	case <-ctx.Done():
-		close(msg.cancel)
 		monitoring.Log().Warn("Process timeout", obj.LogData(zap.String("error", "Context.timeout"))...)
 		return r.replyWithError(obj, 499, errContextCancel)
 	case res := <-msg.responseChan:
@@ -104,9 +101,6 @@ func (r *RequestProcessor) Process(req *http.Request, obj *object.FileObject) *r
 func (r *RequestProcessor) processChan(ctx context.Context, msg requestMessage) {
 	res := r.process(msg.request, msg.obj)
 	select {
-	case <-msg.cancel:
-		res.Close()
-		return
 	case <-ctx.Done():
 		res.Close()
 		return
@@ -243,6 +237,9 @@ func (r *RequestProcessor) collapseGET(req *http.Request, obj *object.FileObject
 			return r.replyWithError(obj, 504, errContextCancel)
 		case res, ok := <-lockResult.ResponseChan:
 			if !ok || res == nil {
+				if cacheRes, err := r.responseCache.Get(obj); err == nil {
+					return cacheRes
+				}
 				return r.handleGET(req, obj)
 			}
 			return res