This is an automated email from the ASF dual-hosted git repository.
flypiggy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-seata-go.git
The following commit(s) were added to refs/heads/master by this push:
new 4a684d6f fix: drain cached fence logs for cleanup task (#1085)
4a684d6f is described below
commit 4a684d6fd969a6f8b69adae9e3e2e158b56a5a6b
Author: waterbucket <[email protected]>
AuthorDate: Sat Apr 11 20:49:39 2026 +0800
fix: drain cached fence logs for cleanup task (#1085)
* fix: drain cached fence logs for cleanup task
* fix: requeue fence logs on drain failure; fix tests and cleanInterval race
* fix: add logCache retry limit and drainCacheTask shutdown test
---------
Co-authored-by: CocaElbow <[email protected]>
Co-authored-by: flypiggyNo3 <[email protected]>
Co-authored-by: ThunGuo <[email protected]>
---
.../tcc/fence/handler/tcc_fence_wrapper_handler.go | 131 ++++++-
.../handler/tcc_fence_wrapper_handler_test.go | 380 ++++++++++++++++++++-
2 files changed, 499 insertions(+), 12 deletions(-)
diff --git a/pkg/rm/tcc/fence/handler/tcc_fence_wrapper_handler.go
b/pkg/rm/tcc/fence/handler/tcc_fence_wrapper_handler.go
index 2a656591..5b159fae 100644
--- a/pkg/rm/tcc/fence/handler/tcc_fence_wrapper_handler.go
+++ b/pkg/rm/tcc/fence/handler/tcc_fence_wrapper_handler.go
@@ -24,6 +24,7 @@ import (
"errors"
"fmt"
"sync"
+ "sync/atomic"
"time"
"github.com/go-sql-driver/mysql"
@@ -40,25 +41,45 @@ type tccFenceWrapperHandler struct {
tccFenceDao dao.TCCFenceStore
logQueue chan *model.FenceLogIdentity
logCache list.List
+ cacheMutex sync.Mutex
logQueueOnce sync.Once
logQueueCloseOnce sync.Once
+ logCacheOnce sync.Once
logTaskOnce sync.Once
db *sql.DB
dbMutex sync.RWMutex
+ stopDrainCache chan struct{}
}
const (
maxQueueSize = 500
channelDelete = 5
cleanExpired = 24 * time.Hour
+ // maxFenceLogCacheRetries is how many times a fence log identity may
be re-queued to logCache
+ // after a failed drain (Begin/delete/commit). Beyond this, entries are
dropped to avoid unbounded retry.
+ maxFenceLogCacheRetries = 3
)
+// fenceLogCacheEntry is the value type stored in logCache (queue-full
overflow path).
+type fenceLogCacheEntry struct {
+ identity model.FenceLogIdentity
+ retryCount int
+}
+
var (
- fenceHandler *tccFenceWrapperHandler
- fenceOnce sync.Once
- cleanInterval = 5 * time.Minute
+ fenceHandler *tccFenceWrapperHandler
+ fenceOnce sync.Once
+ cleanIntervalNanos atomic.Int64
)
+func init() {
+ cleanIntervalNanos.Store(int64(5 * time.Minute))
+}
+
+func currentCleanInterval() time.Duration {
+ return time.Duration(cleanIntervalNanos.Load())
+}
+
func GetFenceHandler() *tccFenceWrapperHandler {
if fenceHandler == nil {
fenceOnce.Do(func() {
@@ -70,8 +91,8 @@ func GetFenceHandler() *tccFenceWrapperHandler {
return fenceHandler
}
-func (handler *tccFenceWrapperHandler) InitCleanPeriod(time time.Duration) {
- cleanInterval = time
+func (handler *tccFenceWrapperHandler) InitCleanPeriod(d time.Duration) {
+ cleanIntervalNanos.Store(int64(d))
}
func (handler *tccFenceWrapperHandler) PrepareFence(ctx context.Context, tx
*sql.Tx) error {
@@ -188,7 +209,7 @@ func (handler *tccFenceWrapperHandler)
InitLogCleanChannel(dsn string) {
func (handler *tccFenceWrapperHandler) initLogCleanTask(db *sql.DB) {
- ticker := time.NewTicker(cleanInterval)
+ ticker := time.NewTicker(currentCleanInterval())
defer ticker.Stop()
for range ticker.C {
@@ -225,6 +246,9 @@ func (handler *tccFenceWrapperHandler)
enqueueFenceLogIdentities(identityList []
func (handler *tccFenceWrapperHandler) DestroyLogCleanChannel() {
handler.logQueueCloseOnce.Do(func() {
close(handler.logQueue)
+ if handler.stopDrainCache != nil {
+ close(handler.stopDrainCache)
+ }
handler.dbMutex.Lock()
if handler.db != nil {
handler.db.Close()
@@ -243,16 +267,21 @@ func (handler *tccFenceWrapperHandler)
deleteBatchFence(tx *sql.Tx, batch []mode
}
func (handler *tccFenceWrapperHandler) pushCleanChannel(xid string, branchId
int64) {
- // todo implement
fli := &model.FenceLogIdentity{
Xid: xid,
BranchId: branchId,
}
select {
case handler.logQueue <- fli:
- // todo add batch delete from log cache.
default:
- handler.logCache.PushBack(fli)
+ handler.cacheMutex.Lock()
+ handler.logCache.PushBack(&fenceLogCacheEntry{identity: *fli})
+ handler.cacheMutex.Unlock()
+
+ handler.logCacheOnce.Do(func() {
+ handler.stopDrainCache = make(chan struct{})
+ go handler.drainCacheTask()
+ })
}
log.Infof("add one log to clean queue: %v ", fli)
}
@@ -293,3 +322,87 @@ func (handler *tccFenceWrapperHandler)
traversalCleanChannel(db *sql.DB) {
}
}
}
+
+func (handler *tccFenceWrapperHandler) drainCacheTask() {
+ ticker := time.NewTicker(currentCleanInterval())
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ticker.C:
+ handler.cacheMutex.Lock()
+
+ if handler.logCache.Len() == 0 {
+ handler.cacheMutex.Unlock()
+ continue
+ }
+
+ handler.dbMutex.RLock()
+ db := handler.db
+ handler.dbMutex.RUnlock()
+ if db == nil {
+ handler.cacheMutex.Unlock()
+ continue
+ }
+
+ var drained []fenceLogCacheEntry
+ for e := handler.logCache.Front(); e != nil; {
+ next := e.Next()
+
+ ent := e.Value.(*fenceLogCacheEntry)
+ drained = append(drained, *ent)
+ handler.logCache.Remove(e)
+
+ e = next
+ }
+
+ handler.cacheMutex.Unlock()
+
+ batch := make([]model.FenceLogIdentity, len(drained))
+ for i := range drained {
+ batch[i] = drained[i].identity
+ }
+
+ if len(batch) == 0 {
+ continue
+ }
+ requeue := func() {
+ handler.cacheMutex.Lock()
+ for _, it := range drained {
+ if it.retryCount >=
maxFenceLogCacheRetries {
+ log.Errorf("max fence log cache
retries exceeded, dropping: xid=%s, branchId=%d",
+ it.identity.Xid,
it.identity.BranchId)
+ continue
+ }
+
handler.logCache.PushBack(&fenceLogCacheEntry{
+ identity: it.identity,
+ retryCount: it.retryCount + 1,
+ })
+ }
+ handler.cacheMutex.Unlock()
+ }
+ tx, err := db.Begin()
+ if err != nil {
+ log.Warnf("failed to begin transaction: %v",
err)
+ requeue()
+ continue
+ }
+ err = handler.deleteBatchFence(tx, batch)
+ if err != nil {
+ _ = tx.Rollback()
+ log.Errorf("delete batch fence log failed,
batch: %v, err: %v", batch, err)
+ requeue()
+ continue
+ }
+ if err = tx.Commit(); err != nil {
+ _ = tx.Rollback()
+ log.Errorf("failed to commit transaction: %v",
err)
+ requeue()
+ continue
+ }
+ case <-handler.stopDrainCache:
+ return
+ }
+ }
+
+}
diff --git a/pkg/rm/tcc/fence/handler/tcc_fence_wrapper_handler_test.go
b/pkg/rm/tcc/fence/handler/tcc_fence_wrapper_handler_test.go
index bc03f747..ea93c4ce 100644
--- a/pkg/rm/tcc/fence/handler/tcc_fence_wrapper_handler_test.go
+++ b/pkg/rm/tcc/fence/handler/tcc_fence_wrapper_handler_test.go
@@ -22,6 +22,7 @@ import (
"database/sql"
"errors"
"sync"
+ "sync/atomic"
"testing"
"time"
@@ -143,10 +144,10 @@ func TestInitCleanPeriod(t *testing.T) {
testDuration := 10 * time.Minute
handler.InitCleanPeriod(testDuration)
- assert.Equal(t, testDuration, cleanInterval)
+ assert.Equal(t, testDuration, currentCleanInterval())
// Reset to default for other tests
- cleanInterval = 5 * time.Minute
+ cleanIntervalNanos.Store(int64(5 * time.Minute))
}
func TestPrepareFence_Success(t *testing.T) {
@@ -796,6 +797,8 @@ func TestPushCleanChannel_FullQueue(t *testing.T) {
logQueue: make(chan *model.FenceLogIdentity, 1),
}
+ // Second pushCleanChannel starts drainCacheTask; stop it before the
next test mutates cleanInterval.
+ defer handler.DestroyLogCleanChannel()
// Fill the queue
handler.pushCleanChannel("xid1", 1)
@@ -1062,5 +1065,376 @@ func TestConstants(t *testing.T) {
assert.Equal(t, 500, maxQueueSize)
assert.Equal(t, 5, channelDelete)
assert.Equal(t, 24*time.Hour, cleanExpired)
- assert.Equal(t, 5*time.Minute, cleanInterval)
+ assert.Equal(t, 5*time.Minute, currentCleanInterval())
+}
+
+func TestDrainCacheTask(t *testing.T) {
+ log.Init()
+
+ oldInterval := currentCleanInterval()
+ cleanIntervalNanos.Store(int64(20 * time.Millisecond))
+ defer cleanIntervalNanos.Store(int64(oldInterval))
+
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+
+ // drainCacheTask should execute one delete batch with Begin + Commit.
+ mock.ExpectBegin()
+ mock.ExpectCommit()
+
+ deleted := make(chan struct{}, 1)
+
+ mockDao := &mockTCCFenceStore{
+ deleteMultipleFunc: func(tx *sql.Tx, identity
[]model.FenceLogIdentity) error {
+ for _, it := range identity {
+ if it.Xid == "xid-cache" && it.BranchId == 2 {
+ select {
+ case deleted <- struct{}{}:
+ default:
+ }
+ }
+ }
+ return nil
+ },
+ }
+
+ handler := &tccFenceWrapperHandler{
+ tccFenceDao: mockDao,
+ logQueue: make(chan *model.FenceLogIdentity, 1), //
Intentionally small queue to trigger the default branch.
+ db: db,
+ }
+
+ // Fill the queue first.
+ handler.pushCleanChannel("xid-queue", 1)
+ // Push one more item so it falls back to cache and starts
drainCacheTask via logCacheOnce.Do(...).
+ handler.pushCleanChannel("xid-cache", 2)
+
+ time.Sleep(100 * time.Millisecond)
+ assert.NotNil(t, handler.stopDrainCache)
+
+ select {
+ case <-deleted:
+ // success
+ case <-time.After(2 * time.Second):
+ t.Fatal("expected drainCacheTask to delete cached identity")
+ }
+
+ assert.NoError(t, mock.ExpectationsWereMet())
+
+ // Cleanup to avoid goroutine leaks.
+ handler.DestroyLogCleanChannel()
+}
+
+func TestDrainCacheTask_DBNil(t *testing.T) {
+ log.Init()
+
+ oldInterval := currentCleanInterval()
+ cleanIntervalNanos.Store(int64(20 * time.Millisecond))
+ defer cleanIntervalNanos.Store(int64(oldInterval))
+
+ handler := &tccFenceWrapperHandler{
+ tccFenceDao: &mockTCCFenceStore{},
+ logQueue: make(chan *model.FenceLogIdentity, 1),
+ stopDrainCache: make(chan struct{}),
+ // Keep db as nil intentionally.
+ }
+
+ handler.cacheMutex.Lock()
+ handler.logCache.PushBack(&fenceLogCacheEntry{identity:
model.FenceLogIdentity{Xid: "xid-nil", BranchId: 9}})
+ handler.cacheMutex.Unlock()
+
+ go handler.drainCacheTask()
+
+ time.Sleep(100 * time.Millisecond)
+
+ handler.cacheMutex.Lock()
+ cacheLen := handler.logCache.Len()
+ handler.cacheMutex.Unlock()
+
+ assert.Equal(t, 1, cacheLen)
+
+ handler.DestroyLogCleanChannel()
+}
+
+func TestDrainCacheTask_DeleteError_RequeuesToCache(t *testing.T) {
+ log.Init()
+
+ oldInterval := currentCleanInterval()
+ cleanIntervalNanos.Store(int64(20 * time.Millisecond))
+ defer cleanIntervalNanos.Store(int64(oldInterval))
+
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+
+ // Each drain tick: Begin -> deleteBatchFence (error) -> Rollback.
Allow several ticks before Destroy stops the goroutine.
+ for i := 0; i < 10; i++ {
+ mock.ExpectBegin()
+ mock.ExpectRollback()
+ }
+
+ var deleteCalls atomic.Int32
+ mockDao := &mockTCCFenceStore{
+ deleteMultipleFunc: func(tx *sql.Tx, identity
[]model.FenceLogIdentity) error {
+ deleteCalls.Add(1)
+ return errors.New("simulated delete failure")
+ },
+ }
+
+ handler := &tccFenceWrapperHandler{
+ tccFenceDao: mockDao,
+ logQueue: make(chan *model.FenceLogIdentity, 1),
+ db: db,
+ }
+
+ handler.pushCleanChannel("xid-queue", 1)
+ handler.pushCleanChannel("xid-cache", 2)
+
+ // Require both in one poll: (1) Len()>=1 alone is true before the
first drain (second push left an item).
+ // (2) deleteCalls>=1 alone can race the next ticker tick (cache
emptied again before we read Len).
+ assert.Eventually(t, func() bool {
+ if deleteCalls.Load() < 1 {
+ return false
+ }
+ handler.cacheMutex.Lock()
+ n := handler.logCache.Len()
+ handler.cacheMutex.Unlock()
+ return n >= 1
+ }, 2*time.Second, 10*time.Millisecond,
+ "delete must have run and logCache must still hold requeued
identities (stable window)")
+
+ handler.DestroyLogCleanChannel()
+}
+
+func TestDrainCacheTask_BeginError_RequeuesToCache(t *testing.T) {
+ log.Init()
+
+ oldInterval := currentCleanInterval()
+ cleanIntervalNanos.Store(int64(20 * time.Millisecond))
+ defer cleanIntervalNanos.Store(int64(oldInterval))
+
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+
+ mock.ExpectBegin().WillReturnError(errors.New("simulated begin
failure"))
+ for i := 0; i < 15; i++ {
+ mock.ExpectBegin()
+ mock.ExpectCommit()
+ }
+
+ var deleteCalls atomic.Int32
+ mockDao := &mockTCCFenceStore{
+ deleteMultipleFunc: func(tx *sql.Tx, identity
[]model.FenceLogIdentity) error {
+ deleteCalls.Add(1)
+ return nil
+ },
+ }
+
+ handler := &tccFenceWrapperHandler{
+ tccFenceDao: mockDao,
+ logQueue: make(chan *model.FenceLogIdentity, 1),
+ db: db,
+ }
+
+ handler.pushCleanChannel("xid-queue", 1)
+ handler.pushCleanChannel("xid-cache", 2)
+
+ // First tick: Begin fails (requeue). deleteCalls stays 0. Later tick:
delete succeeds and cache drains.
+ assert.Eventually(t, func() bool {
+ if deleteCalls.Load() < 1 {
+ return false
+ }
+ handler.cacheMutex.Lock()
+ n := handler.logCache.Len()
+ handler.cacheMutex.Unlock()
+ return n == 0
+ }, 2*time.Second, 10*time.Millisecond,
+ "successful delete after begin failure should empty requeued
cache")
+
+ handler.DestroyLogCleanChannel()
+}
+
+func TestDrainCacheTask_CommitError_RequeuesToCache(t *testing.T) {
+ log.Init()
+
+ oldInterval := currentCleanInterval()
+ cleanIntervalNanos.Store(int64(20 * time.Millisecond))
+ defer cleanIntervalNanos.Store(int64(oldInterval))
+
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+
+ mock.ExpectBegin()
+ mock.ExpectCommit().WillReturnError(errors.New("simulated commit
failure"))
+ for i := 0; i < 15; i++ {
+ mock.ExpectBegin()
+ mock.ExpectCommit()
+ }
+
+ var deleteCalls atomic.Int32
+ mockDao := &mockTCCFenceStore{
+ deleteMultipleFunc: func(tx *sql.Tx, identity
[]model.FenceLogIdentity) error {
+ deleteCalls.Add(1)
+ return nil
+ },
+ }
+
+ handler := &tccFenceWrapperHandler{
+ tccFenceDao: mockDao,
+ logQueue: make(chan *model.FenceLogIdentity, 1),
+ db: db,
+ }
+
+ handler.pushCleanChannel("xid-queue", 1)
+ handler.pushCleanChannel("xid-cache", 2)
+
+ // First tick: delete ok, Commit fails (requeue), deleteCalls==1 and
Len==1. Later tick finishes cleanup.
+ assert.Eventually(t, func() bool {
+ if deleteCalls.Load() < 1 {
+ return false
+ }
+ handler.cacheMutex.Lock()
+ n := handler.logCache.Len()
+ handler.cacheMutex.Unlock()
+ return n == 0
+ }, 2*time.Second, 10*time.Millisecond,
+ "successful commit after prior commit failure should empty
requeued cache")
+
+ handler.DestroyLogCleanChannel()
+}
+
+func TestDrainCacheTask_MaxRetries_DropsFromCache(t *testing.T) {
+ log.Init()
+
+ oldInterval := currentCleanInterval()
+ cleanIntervalNanos.Store(int64(20 * time.Millisecond))
+ defer cleanIntervalNanos.Store(int64(oldInterval))
+
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+
+ for i := 0; i < 12; i++ {
+ mock.ExpectBegin()
+ mock.ExpectRollback()
+ }
+
+ var deleteCalls atomic.Int32
+ mockDao := &mockTCCFenceStore{
+ deleteMultipleFunc: func(tx *sql.Tx, identity
[]model.FenceLogIdentity) error {
+ deleteCalls.Add(1)
+ return errors.New("always fail delete")
+ },
+ }
+
+ handler := &tccFenceWrapperHandler{
+ tccFenceDao: mockDao,
+ logQueue: make(chan *model.FenceLogIdentity, 1),
+ db: db,
+ }
+
+ handler.pushCleanChannel("xid-queue", 1)
+ handler.pushCleanChannel("xid-cache", 2)
+
+ // maxFenceLogCacheRetries=3: requeue with counts 1,2,3 then drop on
fourth failure (fourth successful Begin + failed delete).
+ assert.Eventually(t, func() bool {
+ if deleteCalls.Load() < int32(maxFenceLogCacheRetries+1) {
+ return false
+ }
+ handler.cacheMutex.Lock()
+ n := handler.logCache.Len()
+ handler.cacheMutex.Unlock()
+ return n == 0
+ }, 4*time.Second, 10*time.Millisecond,
+ "after max requeues, failed drain should drop identities from
logCache")
+
+ handler.DestroyLogCleanChannel()
+}
+
+func TestDrainCacheTask_StopsOnDestroy(t *testing.T) {
+ log.Init()
+
+ oldInterval := currentCleanInterval()
+ cleanIntervalNanos.Store(int64(20 * time.Millisecond))
+ defer cleanIntervalNanos.Store(int64(oldInterval))
+
+ waitDrainGoroutine := func(t *testing.T, wg *sync.WaitGroup) {
+ t.Helper()
+ done := make(chan struct{})
+ go func() {
+ wg.Wait()
+ close(done)
+ }()
+ select {
+ case <-done:
+ case <-time.After(3 * time.Second):
+ t.Fatal("drainCacheTask goroutine did not exit after
DestroyLogCleanChannel")
+ }
+ }
+
+ t.Run("emptyLogCache", func(t *testing.T) {
+ var wg sync.WaitGroup
+ wg.Add(1)
+
+ handler := &tccFenceWrapperHandler{
+ tccFenceDao: &mockTCCFenceStore{},
+ logQueue: make(chan *model.FenceLogIdentity,
maxQueueSize),
+ }
+ handler.logCacheOnce.Do(func() {
+ handler.stopDrainCache = make(chan struct{})
+ go func() {
+ defer wg.Done()
+ handler.drainCacheTask()
+ }()
+ })
+
+ time.Sleep(30 * time.Millisecond)
+
+ assert.NotPanics(t, func() {
+ handler.DestroyLogCleanChannel()
+ })
+
+ waitDrainGoroutine(t, &wg)
+ })
+
+ t.Run("withPendingCacheEntries", func(t *testing.T) {
+ var wg sync.WaitGroup
+ wg.Add(1)
+
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+ for i := 0; i < 10; i++ {
+ mock.ExpectBegin()
+ mock.ExpectCommit()
+ }
+
+ handler := &tccFenceWrapperHandler{
+ tccFenceDao: &mockTCCFenceStore{},
+ logQueue: make(chan *model.FenceLogIdentity,
maxQueueSize),
+ db: db,
+ }
+ handler.logCacheOnce.Do(func() {
+ handler.stopDrainCache = make(chan struct{})
+ go func() {
+ defer wg.Done()
+ handler.drainCacheTask()
+ }()
+ })
+
+ handler.cacheMutex.Lock()
+ handler.logCache.PushBack(&fenceLogCacheEntry{identity:
model.FenceLogIdentity{Xid: "xid-pending", BranchId: 1}})
+ handler.cacheMutex.Unlock()
+
+ time.Sleep(50 * time.Millisecond)
+
+ assert.NotPanics(t, func() {
+ handler.DestroyLogCleanChannel()
+ })
+
+ waitDrainGoroutine(t, &wg)
+ })
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
