perms-system-server/internal/model/product/lockByCodeTx_audit_test.go

package product

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"perms-system-server/internal/testutil"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/zeromicro/go-zero/core/stores/sqlx"
)

// ---------------------------------------------------------------------------
// 覆盖目标：审计 M-6 的新增 LockByCodeTx —— "在当前事务里 SELECT ... FOR UPDATE 锁住 product 行"。
// 本方法是后续将 SyncPermissions 串行化到每个 product 的基础设施。契约：
//   1) 存在的 code 必须正常返回全字段；
//   2) 不存在的 code 必须返回 sqlx.ErrNotFound 以便调用方 fail-close；
//   3) 行锁必须真实 —— 两个事务对同一 code 并发 FOR UPDATE，
//      先拿到锁的那个必须让后者阻塞到前者 commit/rollback 为止。
// ---------------------------------------------------------------------------

// TC-0809: LockByCodeTx 对存在的 code 返回完整数据。
func TestSysProductModel_LockByCodeTx_Found(t *testing.T) {
	ctx := context.Background()
	conn := testutil.GetTestSqlConn()
	m := newTestModel(t)

	p := newSysProduct()
	res, err := m.Insert(ctx, p)
	require.NoError(t, err)
	id, _ := res.LastInsertId()
	t.Cleanup(func() { testutil.CleanTable(ctx, conn, "`sys_product`", id) })

	var got *SysProduct
	require.NoError(t, m.TransactCtx(ctx, func(c context.Context, session sqlx.Session) error {
		got, err = m.LockByCodeTx(c, session, p.Code)
		return err
	}))
	require.NotNil(t, got)
	assert.Equal(t, id, got.Id)
	assert.Equal(t, p.Code, got.Code)
	assert.Equal(t, p.AppKey, got.AppKey)
	assert.Equal(t, p.Status, got.Status, "锁行时不得过滤禁用态，否则 SyncPermissions 无法为禁用产品正确 fail-close")
}

// TC-0810: LockByCodeTx 对不存在的 code 返回 sqlx.ErrNotFound。
func TestSysProductModel_LockByCodeTx_NotFound(t *testing.T) {
	ctx := context.Background()
	m := newTestModel(t)

	err := m.TransactCtx(ctx, func(c context.Context, session sqlx.Session) error {
		_, e := m.LockByCodeTx(c, session, "definitely_no_such_code_"+testutil.UniqueId())
		return e
	})
	require.Error(t, err)
	assert.True(t, errors.Is(err, sqlx.ErrNotFound),
		"LockByCodeTx 对不存在的 code 必须返回 ErrNotFound，便于上层 fail-close 返回 401/404")
}

// TC-0811: FOR UPDATE 行锁真实生效 —— 两个事务同时尝试锁同一行时，
// 后进者必须被阻塞直到先进者结束事务。
// 测量方式：goroutine A 在 tx 内 Lock 住后 sleep 500ms 再 commit；
//          goroutine B 等 100ms 后也尝试 Lock 同一行，记录耗时。B 的耗时必须≥400ms。
func TestSysProductModel_LockByCodeTx_BlocksConcurrentWriter(t *testing.T) {
	ctx := context.Background()
	conn := testutil.GetTestSqlConn()
	m := newTestModel(t)

	p := newSysProduct()
	res, err := m.Insert(ctx, p)
	require.NoError(t, err)
	id, _ := res.LastInsertId()
	t.Cleanup(func() { testutil.CleanTable(ctx, conn, "`sys_product`", id) })

	var (
		wg             sync.WaitGroup
		aHoldMs        = int64(500)
		bStartDelayMs  = int64(100)
		bElapsedNanos  int64
		aFinishedNanos int64
		aErr, bErr     error
	)

	wg.Add(2)
	go func() {
		defer wg.Done()
		aErr = m.TransactCtx(ctx, func(c context.Context, session sqlx.Session) error {
			if _, e := m.LockByCodeTx(c, session, p.Code); e != nil {
				return e
			}
			// A 拿到锁后故意延时，模拟一段业务处理期。期间 B 必须被阻塞。
			time.Sleep(time.Duration(aHoldMs) * time.Millisecond)
			atomic.StoreInt64(&aFinishedNanos, time.Now().UnixNano())
			return nil
		})
	}()

	go func() {
		defer wg.Done()
		time.Sleep(time.Duration(bStartDelayMs) * time.Millisecond)
		start := time.Now()
		bErr = m.TransactCtx(ctx, func(c context.Context, session sqlx.Session) error {
			_, e := m.LockByCodeTx(c, session, p.Code)
			return e
		})
		atomic.StoreInt64(&bElapsedNanos, time.Since(start).Nanoseconds())
	}()

	wg.Wait()
	require.NoError(t, aErr)
	require.NoError(t, bErr)

	// B 的耗时 ≥ A 的剩余持锁时间。A 持 500ms，B 延 100ms 后入场，
	// 因此 B 被阻塞的时间至少 (500-100)=400ms。给 DB 一点抖动放到 300ms。
	elapsedMs := atomic.LoadInt64(&bElapsedNanos) / int64(time.Millisecond)
	minBlockedMs := int64(300)
	assert.GreaterOrEqualf(t, elapsedMs, minBlockedMs, fmt.Sprintf(
		"B 的 LockByCodeTx 总耗时 %dms 明显低于预期最小阻塞 %dms —— "+
			"意味着 FOR UPDATE 行锁失效，M-6 声称的'按 product 串行化'不成立",
		elapsedMs, minBlockedMs))
}