Gorm Upexpected transaction behavior in race condition

Unexpected transaction behaviour in race condition

In the following code, - routine1 first locks the record using "FOR UPDATE", then sleep 1 second to simulate time consuming operations, at last update the record's status to STATUS_SUCCESS. - routine2 is trigered after routine1 lock the record, and try to update record's record to STATUS_FAIL

package sql_test

import (
    "github.com/stretchr/testify/require"
    "gorm.io/driver/sqlite"
    "gorm.io/gorm"
    "os"
    "sync"
    "testing"
    "time"
)

type TaskType int32

const (
    TASK_ADD TaskType = 0
)

type TaskStatus int32

const (
    STATUS_SUCCESS TaskStatus = 1
    STATUS_FAIL    TaskStatus = 2
)

// Address define the table account
type Task struct {
    gorm.Model
    Type   TaskType
    Status TaskStatus
}

func TestTransactionInRaceCondition(t *testing.T) {
    db, err := gorm.Open(sqlite.Open("test.db"), &gorm.Config{
        SkipDefaultTransaction: false,
    })
    require.NoError(t, err)
    defer os.Remove("test.db")
    var wg sync.WaitGroup
    s1 := make(chan struct{})


    db.AutoMigrate(&Task{})

    record := Task{
        Type: TASK_ADD,
    }
    err = db.Create(&record).Error
    require.NoError(t, err)

    var res Task
    err = db.First(&res).Where("id = ?", 1).Error
    require.NoError(t, err)

    id := res.ID

    wg.Add(2)
    go func() {
        //routine1 lock the record first
        defer wg.Done()
        tx1 := db.Begin()
        defer func() {
            if r := recover(); r != nil {
                tx1.Rollback()
            }
        }()

        if err1 := tx1.Error; err1 != nil {
            t.Logf("routine1 transaction create error")
        }

        var res1 Task
        err1 := tx1.
            Set("gorm:query_option", "FOR UPDATE").
            Model(&Task{}).
            Where("id = ?", id).
            First(&res1).
            Error

        if err1 != nil {
            t.Logf("routine1 lock error")
            tx1.Rollback()
            return
        } else {
            t.Logf("routine1 lock success")
        }

        s1 <- struct{}{} //trig routine2
        time.Sleep(time.Second * 1)

        err1 = tx1.
            Model(&Task{}).
            Where("id = ?", id).
            Update("status", STATUS_SUCCESS).
            Error

        if err1 != nil {
            t.Logf("routine1 update error")
            tx1.Rollback()
            return
        }

        err1 = tx1.Commit().Error
        if err1 != nil {
            t.Logf("rountine1 commit error")
            tx1.Rollback()
            return
        }
        err1 = db.
            Model(&Task{}).
            Where("id = ?", id).
            First(&res1).
            Error

        t.Logf("routine1 status:%v", res1.Status)
    }()

    go func() {
        //routine2
        defer wg.Done()
        <-s1

        t.Logf("routine2 begin")
        tx2 := db.Begin()
        defer func() {
            if r := recover(); r != nil {
                tx2.Rollback()
            }
        }()

        if err2 := tx2.Error; err2 != nil {
            t.Logf("routine1 transaction create error")
        }

        var res2 Task
        err2 := tx2.
            Set("gorm:query_option", "FOR UPDATE").
            Model(&Task{}).
            Where("id = ?", id).
            First(&res2).
            Error

        if err2 != nil {
            t.Logf("routine2 lock error %v", err2.Error())
            tx2.Rollback()
            return
        } else {
            t.Logf("routine2 lock success")
        }

        err2 = tx2.
            Model(&Task{}).
            Where("id = ?", id).
            Update("status", STATUS_FAIL).
            Error

        if err2 != nil {
            t.Logf("routine2 update error %v", err2.Error())
            tx2.Rollback()
            return
        } else {
            t.Logf("routine2 update success")
        }

        err2 = tx2.Commit().Error
        if err2 != nil {
            t.Logf("routine2 commit error %v", err2.Error())
            tx2.Rollback()
            return
        }

        err2 = db.
            Model(&Task{}).
            Where("id = ?", id).
            First(&res2).
            Error
        t.Logf("routine2 status:%v", res2.Status)
    }()

    wg.Wait()

}

Expected answer

routine1 first update the status to STATUS_SUCCESS, after routine1 done, routine2 update the status to STATUS_FAIL.

but the result is
- routine2 overtaken routine1 lock and update first(guess this is because routine1 sleep) - routine1 update fail.

Comment From: a631807682

This is not a problem of gorm, I don't know the lock rules of sqlite, but it is expected in mysql.

Comment From: readygo67

I change the sqlite to mysql like below func TestTransactionInRaceCondition(t *testing.T) { const TESTDB_DSN = "root:12345678@tcp(127.0.0.1:3306)/testdb?charset=utf8mb4&parseTime=True&loc=Local" db, err := gorm.Open(mysql.Open(TESTDB_DSN), &gorm.Config{ SkipDefaultTransaction: false, }) require.NoError(t, err)

var wg sync.WaitGroup
s1 := make(chan struct{})

db.AutoMigrate(&Task{})

record := Task{
    Type: TASK_ADD,
}
err = db.Create(&record).Error
require.NoError(t, err)

var res Task
err = db.First(&res).Where("id = ?", 1).Error
require.NoError(t, err)

id := res.ID

wg.Add(2)
go func() {
    //routine1 lock the record first
    defer wg.Done()
    tx1 := db.Begin()
    defer func() {
        if r := recover(); r != nil {
            tx1.Rollback()
        }
    }()

    if err1 := tx1.Error; err1 != nil {
        t.Logf("routine1 transaction create error")
    }

    var res1 Task
    err1 := tx1.
        Set("gorm:query_option", "FOR UPDATE").
        Model(&Task{}).
        Where("id = ?", id).
        First(&res1).
        Error

    if err1 != nil {
        t.Logf("routine1 lock error")
        tx1.Rollback()
        return
    } else {
        t.Logf("routine1 lock success")
    }

    s1 <- struct{}{} //trig routine2
    time.Sleep(time.Second * 1)

    err1 = tx1.
        Model(&Task{}).
        Where("id = ?", id).
        Update("status", STATUS_SUCCESS).
        Error

    if err1 != nil {
        t.Logf("routine1 update error")
        tx1.Rollback()
        return
    } else {
        t.Logf("routine1 update success")
    }

    err1 = tx1.Commit().Error
    if err1 != nil {
        t.Logf("rountine1 commit error")
        tx1.Rollback()
        return
    }
    err1 = db.
        Model(&Task{}).
        Where("id = ?", id).
        First(&res1).
        Error

    t.Logf("routine1 status:%v", res1.Status)
}()

go func() {
    //routine2
    defer wg.Done()
    <-s1

    t.Logf("routine2 begin")
    tx2 := db.Begin()
    defer func() {
        if r := recover(); r != nil {
            tx2.Rollback()
        }
    }()

    if err2 := tx2.Error; err2 != nil {
        t.Logf("routine1 transaction create error")
    }

    var res2 Task
    err2 := tx2.
        Set("gorm:query_option", "FOR UPDATE").
        Model(&Task{}).
        Where("id = ?", id).
        First(&res2).
        Error

    if err2 != nil {
        t.Logf("routine2 lock error %v", err2.Error())
        tx2.Rollback()
        return
    } else {
        t.Logf("routine2 lock success")
    }

    err2 = tx2.
        Model(&Task{}).
        Where("id = ?", id).
        Update("status", STATUS_FAIL).
        Error

    if err2 != nil {
        t.Logf("routine2 update error %v", err2.Error())
        tx2.Rollback()
        return
    } else {
        t.Logf("routine2 update success")
    }

    err2 = tx2.Commit().Error
    if err2 != nil {
        t.Logf("routine2 commit error %v", err2.Error())
        tx2.Rollback()
        return
    }

    err2 = db.
        Model(&Task{}).
        Where("id = ?", id).
        First(&res2).
        Error
    t.Logf("routine2 status:%v", res2.Status)
}()

wg.Wait()

} `

Unfortunately, the result shows that routine2 overtakes routine1's lock. Anybody knows the exact reason?

Comment From: readygo67

After I change the "Set("gorm:query_option", "FOR UPDATE")" to "Clauses(clause.Locking{Strength: "UPDATE"})." , the result is as expected.

`func TestTransactionInRaceCondition(t *testing.T) { const TESTDB_DSN = "root:12345678@tcp(127.0.0.1:3306)/testdb?charset=utf8mb4&parseTime=True&loc=Local" db, err := gorm.Open(mysql.Open(TESTDB_DSN), &gorm.Config{ SkipDefaultTransaction: false, }) require.NoError(t, err)

var wg sync.WaitGroup
s1 := make(chan struct{})

db.AutoMigrate(&Task{})

record := Task{
    Type: TASK_ADD,
}
err = db.Create(&record).Error
require.NoError(t, err)

var res Task
err = db.First(&res).Error
require.NoError(t, err)

id := res.ID

wg.Add(2)
go func() {
    //routine1 lock the record first
    defer wg.Done()
    tx1 := db.Begin()
    defer func() {
        if r := recover(); r != nil {
            tx1.Rollback()
        }
    }()

    if err1 := tx1.Error; err1 != nil {
        t.Logf("routine1 transaction create error")
    }

    var res1 Task
    err1 := tx1.
        Model(&Task{}).
        Where("id = ?", id).
        Clauses(clause.Locking{Strength: "UPDATE"}).
        First(&res1).
        Error

    if err1 != nil {
        t.Logf("routine1 lock error")
        tx1.Rollback()
        return
    } else {
        t.Logf("routine1 lock success")
    }

    s1 <- struct{}{} //trig routine2
    time.Sleep(time.Second * 1)

    err1 = tx1.
        Model(&Task{}).
        Where("id = ?", id).
        Update("status", STATUS_SUCCESS).
        Error

    if err1 != nil {
        t.Logf("routine1 update error")
        tx1.Rollback()
        return
    } else {
        t.Logf("routine1 update success")
    }

    err1 = tx1.Commit().Error
    if err1 != nil {
        t.Logf("rountine1 commit error")
        tx1.Rollback()
        return
    } else {
        t.Logf("rountine1 commit success")
    }

    err1 = db.
        Model(&Task{}).
        Where("id = ?", id).
        First(&res1).
        Error

    t.Logf("routine1 status:%v", res1.Status)
}()

go func() {
    //routine2
    defer wg.Done()
    <-s1

    t.Logf("routine2 begin")
    tx2 := db.Begin()
    defer func() {
        if r := recover(); r != nil {
            tx2.Rollback()
        }
    }()

    if err2 := tx2.Error; err2 != nil {
        t.Logf("routine1 transaction create error")
    }

    var res2 Task
    err2 := tx2.
        Model(&Task{}).
        Where("id = ?", id).
        Clauses(clause.Locking{Strength: "UPDATE"}).
        First(&res2).
        Error

    if err2 != nil {
        t.Logf("routine2 lock error %v", err2.Error())
        tx2.Rollback()
        return
    } else {
        t.Logf("routine2 lock success")
    }

    err2 = tx2.
        Model(&Task{}).
        Where("id = ?", id).
        Update("status", STATUS_FAIL).
        Error

    if err2 != nil {
        t.Logf("routine2 update error %v", err2.Error())
        tx2.Rollback()
        return
    } else {
        t.Logf("routine2 update success")
    }

    err2 = tx2.Commit().Error
    if err2 != nil {
        t.Logf("routine2 commit error %v", err2.Error())
        tx2.Rollback()
        return
    } else {
        t.Logf("rountine2 commit success")
    }

    err2 = db.
        Model(&Task{}).
        Where("id = ?", id).
        First(&res2).
        Error
    t.Logf("routine2 status:%v", res2.Status)
}()

wg.Wait()

}`

Comment From: katlyn-walsh

FYI for anyone else stumbling across this issue - it looks like this is a known change in behavior that wasn't documented in the release notes: https://github.com/go-gorm/gorm.io/issues/403