Move time source from db layer to PersistenceManager #6646

ribaraka · 2025-01-28T00:47:25Z

Refactored the system so that the PersistenceManager level handles the timestamp

Why?
#6610

How did you test it?
unit tests

Potential risks

Release notes

Documentation Changes

bowenxia · 2025-01-28T23:37:04Z

common/persistence/nosql/nosql_execution_store_util_test.go

@@ -60,7 +60,7 @@ func TestNosqlExecutionStoreUtils(t *testing.T) {
 						Data:     []byte(`[{"Branches":[{"BranchID":"test-branch-id","BeginNodeID":1,"EndNodeID":2}]}]`),
 					},
 				}
-				return store.prepareCreateWorkflowExecutionRequestWithMaps(workflowSnapshot)
+				return store.prepareCreateWorkflowExecutionRequestWithMaps(workflowSnapshot, time.Now())


Have you had any error when running the unit tests? I was thinking to use a constant time stamp in the tests like this so that in the unit tests it will always keep the same

No, there haven't been any errors in the tests with this implementation. However, I refactored it to use a FixedTime variable for consistency.

ribaraka · 2025-01-30T14:49:30Z

@bowenxia added another commit with a refactor of the history storage. If you have time, please take a look. I'll be sending the work in small chunks, little by little. There are a few TODOs— if you could reply to them, I’d really appreciate it.

bowenxia · 2025-01-31T06:27:03Z

common/persistence/nosql/nosqlplugin/cassandra/history_events.go

@@ -54,19 +53,20 @@ func (db *cdb) InsertIntoHistoryTreeAndNode(ctx context.Context, treeRow *nosqlp
 		// Note: for perf, prefer using batch for inserting more than one records
 		batch := db.session.NewBatch(gocql.LoggedBatch).WithContext(ctx)
 		batch.Query(v2templateInsertTree,
-			treeRow.TreeID, treeRow.BranchID, ancs, persistence.UnixNanoToDBTimestamp(treeRow.CreateTimestamp.UnixNano()), treeRow.Info, timeStamp)
+			// TODO: Should the `persistence.UnixNanoToDBTimestamp(treeRow.CreateTimestamp.UnixNano())` logic be handled by `persistenceManager` as well, or should it remain as is?


Let's leave it as is, since this PR is more related to the created_time and last_updated_time timestamps.

bowenxia · 2025-01-31T06:31:30Z

LGTM so far. Lemme know if you are planning to refactor other persistence managers here, or in a new PR.

ribaraka · 2025-01-31T07:51:20Z

LGTM so far. Lemme know if you are planning to refactor other persistence managers here, or in a new PR.

I'm planning to do everything in one PR. I figured it would be more convenient to have the refactor in a single place rather than splitting it up, unless there are any objections.

ribaraka · 2025-02-05T20:52:05Z

@bowenxia, I refactored the task manager—please take a look. I've added some questions marked as TODOs.

bowenxia · 2025-02-06T02:37:56Z

common/persistence/nosql/nosqlplugin/cassandra/tasks.go

-				ttl = &ttlInt // TTL is an integer
-			}
-		}
+		// TODO: no usage of ttl


Let's leave it as-is. Because this is not the purpose of this PR.
If there's something wrong, it would be easier to find root cause.

bowenxia · 2025-02-06T02:38:02Z

common/persistence/nosql/nosqlplugin/cassandra/tasks.go

-		if ttl != nil {
-			t.Expiry = db.timeSrc.Now().Add(time.Duration(*ttl) * time.Second)
-		}
+		// TODO: removing this, because there is no usage of it


Another issue at this low-level layer is that TTL expiry computation is happening here, whereas I understand that we should remove any computation from db layer entirely. Moving this logic to upstream led nowhere, as TTL isn't used there either.

Since the task is to remove timeSrc from db and we need to discuss this matter later, I could replace the current logic with time.Now() and keep the TODO.

bowenxia · 2025-02-06T02:46:54Z

common/persistence/nosql/nosqlplugin/cassandra/tasks.go

@@ -110,7 +110,7 @@ func fromTaskListPartitionConfig(config *persistence.TaskListPartitionConfig) ma
 // InsertTaskList insert a single tasklist row
 // Return TaskOperationConditionFailure if the condition doesn't meet
 func (db *cdb) InsertTaskList(ctx context.Context, row *nosqlplugin.TaskListRow) error {
-	timeStamp := db.timeSrc.Now()
+	timeStamp := row.LastUpdatedTime


insert is a create operation. It would be confused to have lastUpdatedTime. I understand that you might want to reuse the LastUpdateTime in TaskListRow struct, but

This lastUpdatedTime is used for task object, not for a TaskList row in Cassandra. Thus, it uses time.now() in nosql_task_store.

Changing it would break the logic of creating a task obejct, which is not related to what we are doing in this PR.

Maybe add a field like "timestamp" would be a better idea.

I suggest using a single expression, like currentTimestamp, for all functions (insert, update). What do you think?

bowenxia · 2025-02-06T02:47:11Z

common/persistence/nosql/nosqlplugin/cassandra/tasks.go

@@ -282,7 +282,7 @@ func (db *cdb) InsertTasks(
 	domainID := tasklistCondition.DomainID
 	taskListName := tasklistCondition.TaskListName
 	taskListType := tasklistCondition.TaskListType
-	timeStamp := db.timeSrc.Now()
+	timeStamp := tasklistCondition.LastUpdatedTime


same here. Insert is a create operation

ribaraka · 2025-02-07T18:20:44Z

@bowenxia when you have a moment, please take another look—especially at the Queue Manager. It requires additional attention on how to handle it.

bowenxia · 2025-02-10T22:14:38Z

common/persistence/data_store_interfaces.go

@@ -200,13 +200,13 @@ type (
 		EnqueueMessage(ctx context.Context, messagePayload []byte) error
 		ReadMessages(ctx context.Context, lastMessageID int64, maxCount int) ([]*InternalQueueMessage, error)
 		DeleteMessagesBefore(ctx context.Context, messageID int64) error
-		UpdateAckLevel(ctx context.Context, messageID int64, clusterName string) error
+		UpdateAckLevel(ctx context.Context, messageID int64, clusterName string, now time.Time) error


nit:
maybe consider changing those to currentTimeStamp as well to make those more descriptive.

bowenxia · 2025-02-10T22:23:16Z

common/persistence/nosql/nosql_queue_store.go

@@ -100,7 +101,7 @@ func (q *nosqlQueueStore) EnqueueMessage(
 	if err != nil {
 		return err
 	}
-	_, err = q.tryEnqueue(ctx, q.queueType, getNextID(ackLevels, lastMessageID), messagePayload)
+	_, err = q.tryEnqueue(ctx, q.queueType, getNextID(ackLevels, lastMessageID), messagePayload, time.Now())


let's change the signature of newNoSQLQueueStore() to have a currentTimestamp as a parameter, in that case we can pass the time from persistence manager

Sorry, but I don't see how adding the currentTimestamp parameter to newNoSQLQueueStore would allow passing the time from the persistence manager. The newNoSQLQueueStore leads to the factory and then to the bean.
That said, I still was able to resolve the selected code.

However, my main challenge now is avoiding time generation at the store level, which happens within newNoSQLQueueStore, specifically in the createQueueMetadataEntryIfNotExist method. If the current solution is okay, please let me know.

I've updated the PR. Please take another look. Skip the tests for now, as they are pending our decision on the solution for newNoSQLQueueStore

Yes you are right.

Adding timeSrc to nosql_queue_store might be a way. @taylanisikdemir Do you have an opinion? Since it is too hard to add timeSrc in queue_manager and pass it down to nosql_queue_store.

Update:

Sync with Taylan offilne. Let's add timeSrc as a field in to nosql_queue_store, and then create it during the call of its constructor for now. Also, could you mark this part as a to do for future work? Thanks!

bowenxia · 2025-02-10T22:43:05Z

common/persistence/sql/sql_queue_store_test.go

@@ -352,7 +353,7 @@ func TestUpdateAckLevel(t *testing.T) {
 			require.NoError(t, err, "Failed to create sql queue store")

 			tc.mockSetup(mockDB, mockTx)
-			err = store.UpdateAckLevel(context.Background(), 0, tc.clusterName)
+			err = store.UpdateAckLevel(context.Background(), 0, tc.clusterName, time.Now())


Maybe change those to fixedTime as well

bowenxia · 2025-02-10T22:43:11Z

common/persistence/sql/sql_queue_store_test.go

@@ -719,7 +720,7 @@ func TestUpdateDLQAckLevel(t *testing.T) {
 			require.NoError(t, err, "Failed to create sql queue store")

 			tc.mockSetup(mockDB, mockTx)
-			err = store.UpdateDLQAckLevel(context.Background(), 0, tc.clusterName)
+			err = store.UpdateDLQAckLevel(context.Background(), 0, tc.clusterName, time.Now())


bowenxia · 2025-02-10T22:59:42Z

common/persistence/nosql/nosql_queue_store.go

@@ -248,20 +244,21 @@ func (q *nosqlQueueStore) insertInitialQueueMetadataRecord(
 	queueType persistence.QueueType,


Let's add a currentTimestamp here as an extra field

bowenxia · 2025-02-10T23:00:41Z

common/persistence/nosql/nosql_queue_store.go

@@ -114,16 +115,11 @@ func (q *nosqlQueueStore) EnqueueMessageToDLQ(
 		return err
 	}

-	_, err = q.tryEnqueue(ctx, q.getDLQTypeFromQueueType(), lastMessageID+1, messagePayload)
+	_, err = q.tryEnqueue(ctx, q.getDLQTypeFromQueueType(), lastMessageID+1, messagePayload, time.Now())


Let's do the same as you did for UpdateAckLevel and UpdateDLQAckLevel to have a currentTimestamp as an extra parameter

bowenxia

shard storage refactored LGTM

bowenxia · 2025-02-19T18:57:41Z

common/persistence/nosql/nosql_task_store.go

-				RangeID:         initialRangeID,
-				TaskListKind:    request.TaskListKind,
-				AckLevel:        initialAckLevel,
-				LastUpdatedTime: now,


nit: consider change now to be a more descriptive name like currentTimeStamp

bowenxia · 2025-02-19T18:58:26Z

common/persistence/nosql/nosql_task_store.go

-		TaskListName:    request.TaskListInfo.Name,
-		TaskListType:    request.TaskListInfo.TaskType,
-		RangeID:         request.TaskListInfo.RangeID,
-		LastUpdatedTime: now,


bowenxia · 2025-02-19T19:11:49Z

common/persistence/nosql/nosql_execution_store.go

 		ts := []persistence.Task{task}

 		tasks, err := d.prepareReplicationTasksForWorkflowTxn(task.DomainID, rowTypeReplicationWorkflowID, rowTypeReplicationRunID, ts)
 		if err != nil {
 			return err
 		}
+		tasks[i].CurrentTimeStamp = request.CurrentTimeStamp


It is redundant here to add timestamp to each task since we are storing the entire tasklist into Cassandra; also we do have a creationTime for each task.

I am thinking how to pass one timestamp into the method d.db.InsertReplicationTask without changing the signature. @taylanisikdemir Any suggestions? I am thinking to add the timestamp into ctx, but not sure if it is the best way.

There has been update in this block of code (i.e. Refactor peristence layer to support inserting history tasks of new categories #6671). Adjusted the code accordingly.

…NC-12163.pdf

# Conflicts: # common/persistence/nosql/nosqlplugin/cassandra/tasks_test.go

# Conflicts: # common/persistence/nosql/nosql_execution_store.go

bowenxia · 2025-02-24T18:15:33Z

Reviewing new changes

ribaraka · 2025-02-24T23:43:12Z

@bowenxia, I’ve made some minor changes and added one TODO—please take a look.

Looks like I’ve refactored everything required for this task. The Queue store is still a bit tricky though.
I’ve adjusted all related tests accordingly. Unit, integration, and E2E tests are all passing successfully. I also ran a smoke test.

I think it's ready to remove WIP and start to wrap it up 😄

bowenxia · 2025-02-25T04:34:28Z

Hi @ribaraka sounds good. I was too busy to review the PR today. I'll try get back to you tmr or the day after. Thanks for your contribution!

ribaraka requested review from Shaddoll, neil-xie, davidporter-id-au, Groxx, shijiesheng, jakobht, 3vilhamster, sankari165, dkrotx, taylanisikdemir and demirkayaender as code owners January 28, 2025 00:47

ribaraka mentioned this pull request Jan 28, 2025

Move time source from db layer to PersistenceManager for better separation of concerns #6610

Open

bowenxia reviewed Jan 28, 2025

View reviewed changes

ribaraka force-pushed the mv-timeSrc-to-persistManager branch from e99014a to f1bb070 Compare January 30, 2025 14:41

bowenxia reviewed Jan 31, 2025

View reviewed changes

ribaraka force-pushed the mv-timeSrc-to-persistManager branch from 3472e3e to daf8015 Compare February 5, 2025 20:48

ribaraka requested a review from bowenxia February 5, 2025 20:49

bowenxia reviewed Feb 6, 2025

View reviewed changes

ribaraka requested a review from bowenxia February 6, 2025 20:46

bowenxia reviewed Feb 10, 2025

View reviewed changes

ribaraka force-pushed the mv-timeSrc-to-persistManager branch from 4c8cfb7 to 2b03e69 Compare February 12, 2025 18:20

ribaraka requested a review from bowenxia February 14, 2025 07:57

ribaraka force-pushed the mv-timeSrc-to-persistManager branch from d9d0012 to fc3bf14 Compare February 14, 2025 09:25

bowenxia reviewed Feb 19, 2025

View reviewed changes

ribaraka added 11 commits February 20, 2025 21:51

Move time source from db layer to PersistenceManager, according to CD…

b3e32a2

…NC-12163.pdf

history storage refactored

9673024

task storage refactored

cc330d0

minor task fixes

1609564

# Conflicts: # common/persistence/nosql/nosqlplugin/cassandra/tasks_test.go

domain storage refactored

c122c56

queue storage refactored

5b87824

queue: fix comments

c02b18c

cluster config storage refactored

c02fae2

shard storage refactored

225ecce

execution storage refactored

6de03ec

# Conflicts: # common/persistence/nosql/nosql_execution_store.go

remove timeSrc from db layer

274e058

ribaraka force-pushed the mv-timeSrc-to-persistManager branch from ac8f49b to bbcb70d Compare February 24, 2025 15:36

ribaraka requested a review from bowenxia February 24, 2025 15:43

conflict and test fixes

b19b80a

ribaraka force-pushed the mv-timeSrc-to-persistManager branch from bbcb70d to b19b80a Compare February 24, 2025 15:46

fixes in Workflow Executions

0ec186a

ribaraka force-pushed the mv-timeSrc-to-persistManager branch from 8949948 to 0ec186a Compare February 24, 2025 23:14

ribaraka changed the title ~~[WIP] Move time source from db layer to PersistenceManager~~ Move time source from db layer to PersistenceManager Feb 24, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Move time source from db layer to PersistenceManager #6646

Move time source from db layer to PersistenceManager #6646

ribaraka commented Jan 28, 2025

bowenxia Jan 28, 2025 •

edited

Loading

ribaraka Jan 29, 2025

ribaraka commented Jan 30, 2025

bowenxia Jan 31, 2025

bowenxia commented Jan 31, 2025

ribaraka commented Jan 31, 2025

ribaraka commented Feb 5, 2025

bowenxia Feb 6, 2025

bowenxia Feb 6, 2025

ribaraka Feb 6, 2025

bowenxia Feb 6, 2025 •

edited

Loading

ribaraka Feb 6, 2025

bowenxia Feb 6, 2025

ribaraka commented Feb 7, 2025

bowenxia Feb 10, 2025

bowenxia Feb 10, 2025

ribaraka Feb 12, 2025

bowenxia Feb 19, 2025

bowenxia Feb 19, 2025

bowenxia Feb 10, 2025

bowenxia Feb 10, 2025

bowenxia Feb 10, 2025

bowenxia Feb 10, 2025

bowenxia left a comment

bowenxia Feb 19, 2025

bowenxia Feb 19, 2025

bowenxia Feb 19, 2025

ribaraka Feb 24, 2025

bowenxia commented Feb 24, 2025

ribaraka commented Feb 24, 2025

bowenxia commented Feb 25, 2025

		@@ -248,20 +244,21 @@ func (q *nosqlQueueStore) insertInitialQueueMetadataRecord(
		queueType persistence.QueueType,

Move time source from db layer to PersistenceManager #6646

Are you sure you want to change the base?

Move time source from db layer to PersistenceManager #6646

Conversation

ribaraka commented Jan 28, 2025

bowenxia Jan 28, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ribaraka commented Jan 30, 2025

Choose a reason for hiding this comment

bowenxia commented Jan 31, 2025

ribaraka commented Jan 31, 2025

ribaraka commented Feb 5, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bowenxia Feb 6, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ribaraka commented Feb 7, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bowenxia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bowenxia commented Feb 24, 2025

ribaraka commented Feb 24, 2025

bowenxia commented Feb 25, 2025

bowenxia Jan 28, 2025 •

edited

Loading

bowenxia Feb 6, 2025 •

edited

Loading