Skip to content

Replace in-memory matching test TaskManager with SQLite. #11879

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cbsandeep10 wants to merge 13 commits into
base: main
Choose a base branch
from
+735 −700
Open

Replace in-memory matching test TaskManager with SQLite. #11879

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
0ad83ef
Replace in-memory matching test TaskManager with SQLite.
cbsandeep10 Aug 31, 2026
c6e59ac
Stabilize resetBacklogCounter tests for SQLite-backed task store.
cbsandeep10 Sep 1, 2026
dc09480
Fix golangci failures in matching SQLite test store.
cbsandeep10 Sep 1, 2026
0957fd9
Apply go fix range loops in SQLite test TaskManager.
cbsandeep10 Sep 1, 2026
0fc2755
Clarify skipped backlog DB-error tests for ConditionFailed unload/rel…
cbsandeep10 Sep 2, 2026
9b8ca94
Allow approximate backlog under-count in DB-fault reset tests.
cbsandeep10 Sep 4, 2026
7cb9f26
Stop matching SQLite test TaskManager from faking store semantics.
cbsandeep10 Sep 4, 2026
ecc369a
Skip Error logs when ack persist fails because the queue context is c…
cbsandeep10 Sep 4, 2026
0b8c8f2
Allow undercount and SQL rollback logs in matching DB-fault tests.
cbsandeep10 Sep 8, 2026
bc889f5
Rename testTaskManager fault helper to shouldFault.
cbsandeep10 Sep 11, 2026
713a240
Clarify matching persistence tests and relax DB-fault ack assertion
cbsandeep10 Sep 11, 2026
4e5467b
Preserve SQL context cancel and deadline through persistence errors. …
cbsandeep10 Sep 28, 2026
d585b7c
Name matching test-store mutexes instead of embedding them.
cbsandeep10 Sep 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 9 additions & 9 deletions service/matching/ack_manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ func (s *AckManagerTestSuite) SetupTest() {
}

func (s *AckManagerTestSuite) AddingTasksIncreasesBacklogCounter() {
ackMgr := newTestAckMgr(s.logger)
ackMgr := newTestAckMgr(s.T(), s.logger)

ackMgr.addTask(1)
s.Equal(int64(1), ackMgr.getBacklogCountHint())
Expand All @@ -40,7 +40,7 @@ func (s *AckManagerTestSuite) AddingTasksIncreasesBacklogCounter() {
}

func (s *AckManagerTestSuite) CompleteTaskMovesAckLevelUpToGap() {
ackMgr := newTestAckMgr(s.logger)
ackMgr := newTestAckMgr(s.T(), s.logger)

_, err := ackMgr.db.RenewLease(context.Background())
s.NoError(err)
Expand Down Expand Up @@ -68,7 +68,7 @@ func (s *AckManagerTestSuite) CompleteTaskMovesAckLevelUpToGap() {
}

func (s *AckManagerTestSuite) TestAckManager() {
ackMgr := newTestAckMgr(s.logger)
ackMgr := newTestAckMgr(s.T(), s.logger)

_, err := ackMgr.db.RenewLease(context.Background())
s.NoError(err)
Expand Down Expand Up @@ -135,7 +135,7 @@ func (s *AckManagerTestSuite) TestAckManager() {
}

func (s *AckManagerTestSuite) Sort() {
ackMgr := newTestAckMgr(s.logger)
ackMgr := newTestAckMgr(s.T(), s.logger)

_, err := ackMgr.db.RenewLease(context.Background())
s.NoError(err)
Expand Down Expand Up @@ -177,7 +177,7 @@ func (s *AckManagerTestSuite) Sort() {
}

func BenchmarkAckManager_AddTask(b *testing.B) {
ackMgr := newTestAckMgr(log.NewTestLogger())
ackMgr := newTestAckMgr(b, log.NewTestLogger())

tasks := make([]int, 1000)
for i := range tasks {
Expand All @@ -200,7 +200,7 @@ func BenchmarkAckManager_AddTask(b *testing.B) {
}

func BenchmarkAckManager_CompleteTask(b *testing.B) {
ackMgr := newTestAckMgr(log.NewTestLogger())
ackMgr := newTestAckMgr(b, log.NewTestLogger())

tasks := make([]int, 1000)
b.ResetTimer()
Expand All @@ -224,10 +224,10 @@ func BenchmarkAckManager_CompleteTask(b *testing.B) {
}
}

func newTestAckMgr(logger log.Logger) *ackManager {
tm := newTestTaskManager(logger)
func newTestAckMgr(t testing.TB, logger log.Logger) *ackManager {
tm := newTestTaskManager(t, logger)
cfg := NewConfig(dynamicconfig.NewNoopCollection())
f, _ := tqid.NewTaskQueueFamily("", "test-queue")
f, _ := tqid.NewTaskQueueFamily(defaultNamespaceId, "test-queue")
prtn := f.TaskQueue(enumspb.TASK_QUEUE_TYPE_WORKFLOW).NormalPartition(0)
tlCfg := newTaskQueueConfig(prtn.TaskQueue(), cfg, "test-namespace")
db := newTaskQueueDB(tlCfg, tm, UnversionedQueueKey(prtn), logger, metrics.NoopMetricsHandler, false)
Expand Down
10 changes: 4 additions & 6 deletions service/matching/backlog_manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,9 +74,9 @@ func (s *BacklogManagerTestSuite) SetupTest() {
s.controller = gomock.NewController(s.T())
s.logger = testlogger.NewTestLogger(s.T(), testlogger.FailOnAnyUnexpectedError)
if s.fairness {
s.taskMgr = newTestFairTaskManager(s.logger)
s.taskMgr = newTestFairTaskManager(s.T(), s.logger)
} else {
s.taskMgr = newTestTaskManager(s.logger)
s.taskMgr = newTestTaskManager(s.T(), s.logger)
}

// A capture handler discards recordings while no capture is active (see
Expand All @@ -87,7 +87,7 @@ func (s *BacklogManagerTestSuite) SetupTest() {
s.cfgcli = dynamicconfig.NewMemoryClient()
s.cfgcol = dynamicconfig.NewCollection(s.cfgcli, s.logger)

f, _ := tqid.NewTaskQueueFamily("", "test-queue")
f, _ := tqid.NewTaskQueueFamily(defaultNamespaceId, "test-queue")
prtn := f.TaskQueue(enumspb.TASK_QUEUE_TYPE_WORKFLOW).NormalPartition(0)
queue := UnversionedQueueKey(prtn)
tlCfg := newTaskQueueConfig(prtn.TaskQueue(), NewConfig(s.cfgcol), "test-namespace")
Expand Down Expand Up @@ -731,9 +731,7 @@ func (s *BacklogManagerTestSuite) TestSyncState_UnloadsOnOwnershipLoss() {
// simulate another partition stealing and releasing ownership
db := s.blm.getDB()
tqd := s.taskMgr.getQueueDataByKey(db.queue)
tqd.Lock()
tqd.rangeID++
tqd.Unlock()
tqd.bumpRangeID(s.T())

s.Eventually(unloadCalled.Load, time.Second, 100*time.Millisecond)
}
Expand Down
4 changes: 2 additions & 2 deletions service/matching/db.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -752,7 +752,7 @@ func (db *taskQueueDB) CompleteTasksLessThan(
Subqueue: int(subqueue),
Limit: limit,
})
if err != nil {
if err != nil && ctx.Err() == nil {
db.logger.Error("Persistent store operation failure",
tag.StoreOperationCompleteTasksLessThan,
tag.Error(err),
Expand Down Expand Up @@ -783,7 +783,7 @@ func (db *taskQueueDB) CompleteFairTasksLessThan(
Subqueue: int(subqueue),
Limit: limit,
})
if err != nil {
if err != nil && ctx.Err() == nil {
db.logger.Error("Persistent store operation failure",
tag.StoreOperationCompleteTasksLessThan,
tag.Error(err),
Expand Down
Loading
Loading