Merge pull request redis#105 from x-martinez/add-node-info-error

Add node information in error

Author: hjr265

error.go

@@ -1,6 +1,9 @@
 package redsync
 
-import "errors"
+import (
+	"errors"
+	"fmt"
+)
 
 // ErrFailed is the error resulting if Redsync fails to acquire the lock after
 // exhausting all retries.
@@ -9,3 +12,32 @@ var ErrFailed = errors.New("redsync: failed to acquire lock")
 // ErrExtendFailed is the error resulting if Redsync fails to extend the
 // lock.
 var ErrExtendFailed = errors.New("redsync: failed to extend lock")
+
+// ErrTaken happens when the lock is already taken in a quorum on nodes.
+type ErrTaken struct {
+	Nodes []int
+}
+
+func (err ErrTaken) Error() string {
+	return fmt.Sprintf("lock already taken, locked nodes: %v", err.Nodes)
+}
+
+// ErrNodeTaken is the error resulting if the lock is already taken in one of
+// the cluster's nodes
+type ErrNodeTaken struct {
+	Node int
+}
+
+func (err ErrNodeTaken) Error() string {
+	return fmt.Sprintf("node #%d: lock already taken", err.Node)
+}
+
+// A RedisError is an error communicating with one of the Redis nodes.
+type RedisError struct {
+	Node int
+	Err  error
+}
+
+func (err RedisError) Error() string {
+	return fmt.Sprintf("node #%d: %s", err.Node, err.Err)
+}

mutex.go

@@ -95,7 +95,7 @@ func (m *Mutex) LockContext(ctx context.Context) error {
 			m.until = until
 			return nil
 		}
-		_, err = func() (int, error) {
+		func() (int, error) {
 			ctx, cancel := context.WithTimeout(ctx, time.Duration(int64(float64(m.expiry)*m.timeoutFactor)))
 			defer cancel()
 			return m.actOnPoolsAsync(func(pool redis.Pool) (bool, error) {
@@ -252,27 +252,36 @@ func (m *Mutex) touch(ctx context.Context, pool redis.Pool, value string, expiry
 
 func (m *Mutex) actOnPoolsAsync(actFn func(redis.Pool) (bool, error)) (int, error) {
 	type result struct {
+		Node   int
 		Status bool
 		Err    error
 	}
 
 	ch := make(chan result)
-	for _, pool := range m.pools {
-		go func(pool redis.Pool) {
-			r := result{}
+	for node, pool := range m.pools {
+		go func(node int, pool redis.Pool) {
+			r := result{Node: node}
 			r.Status, r.Err = actFn(pool)
 			ch <- r
-		}(pool)
+		}(node, pool)
 	}
 	n := 0
+	var taken []int
 	var err error
 	for range m.pools {
 		r := <-ch
 		if r.Status {
 			n++
 		} else if r.Err != nil {
-			err = multierror.Append(err, r.Err)
+			err = multierror.Append(err, RedisError{Node: r.Node, Err: r.Err})
+		} else {
+			taken = append(taken, r.Node)
+			err = multierror.Append(err, ErrNodeTaken{Node: r.Node})
 		}
 	}
+
+	if len(taken) >= m.quorum {
+		return n, ErrTaken{Nodes: taken}
+	}
 	return n, err
 }

mutex_test.go

@@ -1,6 +1,7 @@
 package redsync
 
 import (
+	"errors"
 	"strconv"
 	"testing"
 	"time"
@@ -33,6 +34,28 @@ func TestMutex(t *testing.T) {
 	}
 }
 
+func TestMutexAlreadyLocked(t *testing.T) {
+	for k, v := range makeCases(4) {
+		t.Run(k, func(t *testing.T) {
+			rs := New(v.pools...)
+			key := "test-lock"
+
+			mutex1 := rs.NewMutex(key)
+			err := mutex1.Lock()
+			if err != nil {
+				t.Fatalf("mutex lock failed: %s", err)
+			}
+			assertAcquired(t, v.pools, mutex1)
+
+			mutex2 := rs.NewMutex(key)
+			err = mutex2.Lock()
+			if !errors.As(err, &ErrTaken{}) {
+				t.Fatalf("mutex was not already locked: %s", err)
+			}
+		})
+	}
+}
+
 func TestMutexExtend(t *testing.T) {
 	for k, v := range makeCases(8) {
 		t.Run(k, func(t *testing.T) {
@@ -82,8 +105,8 @@ func TestMutexExtendExpired(t *testing.T) {
 			time.Sleep(1 * time.Second)
 
 			ok, err := mutex.Extend()
-			if err != nil {
-				t.Fatalf("mutex extend failed: %s", err)
+			if err == nil {
+				t.Fatalf("mutex extend didn't fail")
 			}
 			if ok {
 				t.Fatalf("Expected ok == false, got %v", ok)
@@ -108,8 +131,8 @@ func TestMutexUnlockExpired(t *testing.T) {
 			time.Sleep(1 * time.Second)
 
 			ok, err := mutex.Unlock()
-			if err != nil {
-				t.Fatalf("mutex unlock failed: %s", err)
+			if err == nil {
+				t.Fatalf("mutex unlock didn't fail")
 			}
 			if ok {
 				t.Fatalf("Expected ok == false, got %v", ok)
@@ -136,8 +159,8 @@ func TestMutexQuorum(t *testing.T) {
 					assertAcquired(t, v.pools, mutex)
 				} else {
 					err := mutex.Lock()
-					if err != ErrFailed {
-						t.Fatalf("Expected err == %q, got %q", ErrFailed, err)
+					if errors.Is(err, &ErrNodeTaken{}) {
+						t.Fatalf("Expected err == %q, got %q", ErrNodeTaken{}, err)
 					}
 				}
 			}
@@ -258,7 +281,7 @@ func newTestMutexes(pools []redis.Pool, name string, n int) []*Mutex {
 	mutexes := make([]*Mutex, n)
 	for i := 0; i < n; i++ {
 		mutexes[i] = &Mutex{
-			name:          name,
+			name:          name + "-" + strconv.Itoa(i),
 			expiry:        8 * time.Second,
 			tries:         32,
 			delayFunc:     func(tries int) time.Duration { return 500 * time.Millisecond },