Move to upsert strategy; all tests work

integrations/locked_resource_test.go

@@ -10,15 +10,19 @@ import (
 	"time"
 
 	"code.gitea.io/gitea/models"
+	"code.gitea.io/gitea/modules/log"
 
 	"github.com/stretchr/testify/assert"
 )
 
 const (
-	// Note: these values might require tuning
-	before = 500 * time.Millisecond
-	after = 1000 * time.Millisecond
-	tolerance = 200 * time.Millisecond
+	// The tests will fail if the waiter function takes less than
+	// blockerDelay minus tolerance to complete.
+	// Note: these values might require tuning in order to avoid
+	// false negatives.
+	waiterDelay = 100 * time.Millisecond
+	blockerDelay = 200 * time.Millisecond
+	tolerance = 50 * time.Millisecond	// Should be <= (blockerDelay-waiterDelay)/2
 )
 
 type waitResult struct {
@@ -35,37 +39,31 @@ func TestLockedResource(t *testing.T) {
 	// the more certain we are the second goroutine is waiting.
 
 	// This check **must** fail as we're not blocking anything
-	assert.Error(t, blockTest("no block", func(ctx models.DBContext) (func() error, error){
-		return func() error{
-			return nil
-		}, nil
+	assert.Error(t, blockTest("no block", func(ctx models.DBContext) error {
+		return nil
 	}))
 
 	models.AssertNotExistsBean(t, &models.LockedResource{LockType: "test-1", LockKey: 1})
 
 	// Test with creation (i.e. new lock type)
-	assert.NoError(t, blockTest("block-new", func(ctx models.DBContext) (func() error, error){
+	assert.NoError(t, blockTest("block-new", func(ctx models.DBContext) error {
 		_, err := models.GetLockedResourceCtx(ctx, "block-test-1", 1)
-		return func() error{
-			return nil
-		}, err
+		return err
 	}))
 
 	// Test without creation (i.e. lock type already exists)
-	assert.NoError(t, blockTest("block-existing", func(ctx models.DBContext) (func() error, error){
+	assert.NoError(t, blockTest("block-existing", func(ctx models.DBContext) error {
 		_, err := models.GetLockedResourceCtx(ctx, "block-test-1", 1)
-		return func() error{
-			return nil
-		}, err
+		return err
 	}))
 
 	// Test with temporary record
-	assert.NoError(t, blockTest("block-temp", func(ctx models.DBContext) (func() error, error){
+	assert.NoError(t, blockTest("block-temp", func(ctx models.DBContext) error {
 		return models.TempLockResourceCtx(ctx, "temp-1", 1)
 	}))
 }
 
-func blockTest(name string, f func(ctx models.DBContext) (func() error, error)) error {
+func blockTest(name string, f func(ctx models.DBContext) error) error {
 	cb := make(chan waitResult)
 	cw := make(chan waitResult)
 	ref := time.Now()
@@ -86,40 +84,39 @@ func blockTest(name string, f func(ctx models.DBContext) (func() error, error))
 		return resw.Err
 	}
 
-	if resw.Waited < after - tolerance {
+	if resw.Waited < blockerDelay - tolerance {
 		return fmt.Errorf("Waiter not blocked on %s; wait: %d ms, expected > %d ms",
-			name, resw.Waited.Milliseconds(), (after - tolerance).Milliseconds())
+			name, resw.Waited.Milliseconds(), (blockerDelay - tolerance).Milliseconds())
 	}
 
 	return nil
 }
 
-func blockTestFunc(name string, blocker bool, ref time.Time, f func(ctx models.DBContext) (func() error, error)) (wr waitResult) {
+func blockTestFunc(name string, blocker bool, ref time.Time, f func(ctx models.DBContext) error) (wr waitResult) {
 	if blocker {
 		name = fmt.Sprintf("blocker [%s]", name)
 	} else {
 		name = fmt.Sprintf("waiter [%s]", name)
 	}
 	err := models.WithTx(func(ctx models.DBContext) error {
-		fmt.Printf("Entering %s @%d\n", name, time.Now().Sub(ref).Milliseconds())
+		log.Trace("Entering %s @%d", name, time.Now().Sub(ref).Milliseconds())
 		if !blocker {
-			fmt.Printf("Waiting on %s @%d\n", name, time.Now().Sub(ref).Milliseconds())
-			time.Sleep(before)
-			fmt.Printf("Wait finished on %s @%d\n", name, time.Now().Sub(ref).Milliseconds())
+			log.Trace("Waiting on %s @%d", name, time.Now().Sub(ref).Milliseconds())
+			time.Sleep(waiterDelay)
+			log.Trace("Wait finished on %s @%d", name, time.Now().Sub(ref).Milliseconds())
 		}
-		releaseLock, err := f(ctx)
-		if err != nil {
+		if err := f(ctx); err != nil {
 			return err
 		}
 		if blocker {
-			fmt.Printf("Waiting on %s @%d\n", name, time.Now().Sub(ref).Milliseconds())
-			time.Sleep(after)
-			fmt.Printf("Wait finished on %s @%d\n", name, time.Now().Sub(ref).Milliseconds())
+			log.Trace("Waiting on %s @%d", name, time.Now().Sub(ref).Milliseconds())
+			time.Sleep(blockerDelay)
+			log.Trace("Wait finished on %s @%d", name, time.Now().Sub(ref).Milliseconds())
 		} else {
 			wr.Waited = time.Now().Sub(ref)
 		}
-		fmt.Printf("Finishing %s @%d\n", name, time.Now().Sub(ref).Milliseconds())
-		return releaseLock()
+		log.Trace("Finishing %s @%d", name, time.Now().Sub(ref).Milliseconds())
+		return nil
 	})
 	if err != nil {
 		wr.Err = fmt.Errorf("error in %s: %v", name, err)

models/locked_resource.go

@@ -6,41 +6,30 @@ package models
 
 import (
 	"fmt"
+
+	"code.gitea.io/gitea/modules/setting"
 )
 
 type LockedResource struct {
-	LockType	string      `xorm:"pk"`
+	LockType	string      `xorm:"pk VARCHAR(30)"`
 	LockKey		int64		`xorm:"pk"`
 	Counter		int64		`xorm:"NOT NULL DEFAULT 0"`
 }
 
 func GetLockedResource(e Engine, lockType string, lockKey int64) (*LockedResource, error) {
 	locked := &LockedResource{LockType: lockType, LockKey: lockKey}
-	// SQLite3 has no ForUpdate() clause and an UPSERT strategy has many
-	// problems and fallbacks; we perform a bogus update on the table
-	// which will lock the key in a safe way.
-	// Make sure to leave `counter` out of the update.
-	count, err := e.Table(locked).Cols("lock_type", "lock_key").Update(locked)
-	if err != nil {
-		return nil, fmt.Errorf("get locked resource %s:%d: %v", lockType, lockKey, err)
+
+	if err := upsertLockedResource(e, locked); err != nil {
+		return nil, fmt.Errorf("upsertLockedResource: %v", err)
 	}
-	if count == 0 {
-		// No record was found; since the key is now locked,
-		// it's safe to insert a record.
-		_, err = e.Insert(locked)
-		if err != nil {
-			return nil, fmt.Errorf("get locked resource %s:%d: %v", lockType, lockKey, err)
-		}
-	} else {
-		// Read back the record we've locked
-		has, err := e.Table(locked).Get(locked)
-		if err != nil {
-			return nil, fmt.Errorf("get locked resource %s:%d: %v", lockType, lockKey, err)
-		}
-		if !has {
-			return nil, fmt.Errorf("get locked resource %s:%d: record not found", lockType, lockKey)
-		}
+
+	// Read back the record we've created or locked to get the current Counter value
+	if has, err := e.Table(locked).Get(locked); err != nil {
+		return nil, fmt.Errorf("get locked resource %s:%d: %v", lockType, lockKey, err)
+	} else if !has {
+		return nil, fmt.Errorf("unexpected upsert fail  %s:%d", lockType, lockKey)
 	}
+
 	return locked, nil
 }
 
@@ -54,20 +43,15 @@ func DeleteLockedResource(e Engine, resource *LockedResource) error {
 	return err
 }
 
-func TempLockResource(e Engine, lockType string, lockKey int64) (func() error, error) {
+func TempLockResource(e Engine, lockType string, lockKey int64) error {
 	locked := &LockedResource{LockType: lockType, LockKey: lockKey}
-	// Temporary locked resources must not exist in the table
+	// Temporary locked resources must not exist in the table.
+	// This allows us to use a simple INSERT to lock the key.
 	_, err := e.Insert(locked)
-	if err != nil {
-		return func() error {
-			return nil
-		},
-		fmt.Errorf("insert locked resource %s:%d: %v", lockType, lockKey, err)
+	if err == nil {
+		_, err = e.Delete(locked)
 	}
-	return func() error {
-		_, err := e.Delete(locked)
-		return err
-	}, nil
+	return err
 }
 
 func GetLockedResourceCtx(ctx DBContext, lockType string, lockKey int64) (*LockedResource, error) {
@@ -82,7 +66,33 @@ func DeleteLockedResourceCtx(ctx DBContext, resource *LockedResource) error {
 	return DeleteLockedResource(ctx.e, resource)
 }
 
-func TempLockResourceCtx(ctx DBContext, lockType string, lockKey int64) (func() error, error) {
+func TempLockResourceCtx(ctx DBContext, lockType string, lockKey int64) error {
 	return TempLockResource(ctx.e, lockType, lockKey)
 }
 
+func upsertLockedResource(e Engine, resource *LockedResource) (err error) {
+	// An atomic UPSERT operation (INSERT/UPDATE) is the only operation
+	// that ensures that the key is actually locked. 
+	switch {
+	case setting.Database.UseSQLite3 || setting.Database.UsePostgreSQL:
+		_, err = e.Exec("INSERT INTO locked_resource (lock_type, lock_key) "+
+			"VALUES (?,?) ON CONFLICT(lock_type, lock_key) DO UPDATE SET lock_key = ?",
+			resource.LockType, resource.LockKey, resource.LockKey);
+	case setting.Database.UseMySQL:
+		_, err = e.Exec("INSERT INTO locked_resource (lock_type, lock_key) "+
+			"VALUES (?,?) ON DUPLICATE KEY UPDATE lock_key = lock_key",
+			resource.LockType, resource.LockKey);
+	case setting.Database.UseMSSQL:
+		// https://weblogs.sqlteam.com/dang/2009/01/31/upsert-race-condition-with-merge/
+		_, err = e.Exec("MERGE locked_resource WITH (HOLDLOCK) as target "+
+			"USING (SELECT ? AS lock_type, ? AS lock_key) AS src "+
+			"ON src.lock_type = target.lock_type AND src.lock_key = target.lock_key "+
+			"WHEN MATCHED THEN UPDATE SET target.lock_key = target.lock_key "+
+			"WHEN NOT MATCHED THEN INSERT (lock_type, lock_key) "+
+			"VALUES (src.lock_type, src.lock_key);",
+			resource.LockType, resource.LockKey);
+	default:
+		return fmt.Errorf("database type not supported")
+	}
+	return
+}

models/locked_resource_test.go

@@ -47,7 +47,8 @@ func TestLockedResource(t *testing.T) {
 
 	// Attempt temp lock on an existing key, expect error
 	withSession(t, func(t *testing.T, sess *xorm.Session) bool {
-		_, err := TempLockResource(sess, "test-1",1)
+		err := TempLockResource(sess, "test-1",1)
+		// Must give error
 		return assert.Error(t, err)
 	})
 
@@ -62,15 +63,10 @@ func TestLockedResource(t *testing.T) {
 
 	// Attempt temp lock on an valid key, expect success
 	withSession(t, func(t *testing.T, sess *xorm.Session) bool {
-		// Must give error
-		releaseLock, err := TempLockResource(sess, "test-1",1)
-		if !assert.NoError(t, err) {
-			return false
-		}
-		return assert.NoError(t, releaseLock())
+		return assert.NoError(t, TempLockResource(sess, "test-1",1))
 	})
 
 	// Note: testing the validity of the locking mechanism (i.e. whether it actually locks)
-	// must be done in the integration tests, so all the supported databases are checked.
+	// is be done at the integration tests to ensure that all the supported databases are checked.
 }
 

models/migrations/v125.go

@@ -11,7 +11,7 @@ import (
 func addLockedResourceTable(x *xorm.Engine) error {
 
 	type LockedResource struct {
-		LockType	string      `xorm:"pk"`
+		LockType	string      `xorm:"pk VARCHAR(30)"`
 		LockKey		int64		`xorm:"pk"`
 		Counter		int64		`xorm:"NOT NULL DEFAULT 0"`
 	}