Merge pull request rapidpro#600 from nyaruka/locker_cleanup

Replace last usages of old locker code

core/models/contacts.go

@@ -19,6 +19,7 @@ import (
 	"github.com/nyaruka/goflow/excellent/types"
 	"github.com/nyaruka/goflow/flows"
 	"github.com/nyaruka/null"
+	"github.com/nyaruka/redisx"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
@@ -1341,9 +1342,10 @@ func (i *ContactID) Scan(value interface{}) error {
 	return null.ScanInt(value, (*null.Int)(i))
 }
 
-// ContactLock returns the lock key for a particular contact, used with locker
-func ContactLock(orgID OrgID, contactID ContactID) string {
-	return fmt.Sprintf("c:%d:%d", orgID, contactID)
+// GetContactLocker returns the locker for a particular contact
+func GetContactLocker(orgID OrgID, contactID ContactID) *redisx.Locker {
+	key := fmt.Sprintf("lock:c:%d:%d", orgID, contactID)
+	return redisx.NewLocker(key, time.Minute*5)
 }
 
 // UpdateContactModifiedBy updates modified by the passed user id on the passed in contacts

core/runner/runner.go

@@ -15,7 +15,7 @@ import (
 	"github.com/nyaruka/mailroom/core/models"
 	"github.com/nyaruka/mailroom/core/queue"
 	"github.com/nyaruka/mailroom/runtime"
-	"github.com/nyaruka/mailroom/utils/locker"
+	"github.com/nyaruka/redisx"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
@@ -461,7 +461,7 @@ func StartFlow(
 	start := time.Now()
 
 	// map of locks we've released
-	released := make(map[string]bool)
+	released := make(map[*redisx.Locker]bool)
 
 	for len(remaining) > 0 && time.Since(start) < time.Minute*5 {
 		locked := make([]models.ContactID, 0, len(remaining))
@@ -470,8 +470,9 @@ func StartFlow(
 
 		// try up to a second to get a lock for a contact
 		for _, contactID := range remaining {
-			lockID := models.ContactLock(oa.OrgID(), contactID)
-			lock, err := locker.GrabLock(rt.RP, lockID, time.Minute*5, time.Second)
+			locker := models.GetContactLocker(oa.OrgID(), contactID)
+
+			lock, err := locker.Grab(rt.RP, time.Second)
 			if err != nil {
 				return nil, errors.Wrapf(err, "error attempting to grab lock")
 			}
@@ -484,8 +485,8 @@ func StartFlow(
 
 			// defer unlocking if we exit due to error
 			defer func() {
-				if !released[lockID] {
-					locker.ReleaseLock(rt.RP, lockID, lock)
+				if !released[locker] {
+					locker.Release(rt.RP, lock)
 				}
 			}()
 		}
@@ -517,9 +518,9 @@ func StartFlow(
 
 		// release all our locks
 		for i := range locked {
-			lockID := models.ContactLock(oa.OrgID(), locked[i])
-			locker.ReleaseLock(rt.RP, lockID, locks[i])
-			released[lockID] = true
+			locker := models.GetContactLocker(oa.OrgID(), locked[i])
+			locker.Release(rt.RP, locks[i])
+			released[locker] = true
 		}
 
 		// skipped are now our remaining

core/tasks/handler/worker.go

@@ -22,7 +22,6 @@ import (
 	"github.com/nyaruka/mailroom/core/queue"
 	"github.com/nyaruka/mailroom/core/runner"
 	"github.com/nyaruka/mailroom/runtime"
-	"github.com/nyaruka/mailroom/utils/locker"
 	"github.com/nyaruka/null"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -61,8 +60,9 @@ func handleContactEvent(ctx context.Context, rt *runtime.Runtime, task *queue.Ta
 	}
 
 	// acquire the lock for this contact
-	lockID := models.ContactLock(models.OrgID(task.OrgID), eventTask.ContactID)
-	lock, err := locker.GrabLock(rt.RP, lockID, time.Minute*5, time.Second*10)
+	locker := models.GetContactLocker(models.OrgID(task.OrgID), eventTask.ContactID)
+
+	lock, err := locker.Grab(rt.RP, time.Second*10)
 	if err != nil {
 		return errors.Wrapf(err, "error acquiring lock for contact %d", eventTask.ContactID)
 	}
@@ -81,7 +81,7 @@ func handleContactEvent(ctx context.Context, rt *runtime.Runtime, task *queue.Ta
 		}).Info("failed to get lock for contact, requeued and skipping")
 		return nil
 	}
-	defer locker.ReleaseLock(rt.RP, lockID, lock)
+	defer locker.Release(rt.RP, lock)
 
 	// read all the events for this contact, one by one
 	contactQ := fmt.Sprintf("c:%d:%d", task.OrgID, eventTask.ContactID)