fix: use next_event_id column as source of truth when reading workflow execution from Cassandra (#7738)

<!-- 1-2 line summary of WHAT changed technically:
- Always link the relevant projects GitHub issue, unless it is a minor
bugfix
- Good: "Modified FailoverDomain mapper to allow null ActiveClusterName
#320"
- Bad: "added nil check" -->
**What changed?**
Read from the denormalized columns next_event_id (that duplicate the
data next_event_id from the execution field) and set to
InternalWorkflowExecutionInfo when reading concrete execution data from
Cassandra.

<!-- Your goal is to provide all the required context for a future
maintainer
to understand the reasons for making this change (see
https://cbea.ms/git-commit/#why-not-how).
How did this work previously (and what was wrong with it)? What has
changed, and why did you solve it
this way?
- Good: "Active-active domains have independent cluster attributes per
region. Previously,
modifying cluster attributes required spedifying the default
ActiveClusterName which
updates the global domain default. This prevents operators from updating
regional
configurations without affecting the primary cluster designation. This
change allows
  attribute updates to be independent of active cluster selection."
- Bad: "Improves domain handling" -->
**Why?**
Cassandra stores values from the same row but different columns in
different places on disk, rather than as a single, contiguous row block.
It's possible that the denormalized columns get out of sync with the
execution blob in the execution field. This denormalized column is used
as conditional write when updating the execution record for concrete
workflow executions.

By reading it and setting it on InternalWorkflowExecutionInfo we can
leverage the checksum verification to detect differences between the
denormalized next_event_id column and the next_event_id in the execution
blob field (used to calculate the checksum) and identify corrupt
workflows quicker and with more precision.

<!-- Include specific test commands and setup. Please include the exact
commands such that
another maintainer or contributor can reproduce the test steps taken. 
- e.g Unit test commands with exact invocation
`go test -v ./common/types/mapper/proto -run TestFailoverDomainRequest`
- For integration tests include setup steps and test commands
Example: "Started local server with `./cadence start`, then ran `make
test_e2e`"
- For local simulation testing include setup steps for the server and
how you ran the tests
- Good: Full commands that reviewers can copy-paste to verify
- Bad: "Tested locally" or "Added tests" -->
**How did you test it?**
go test ./common/persistence/nosql/nosqlplugin/cassandra -run
Test_parseWorkflowExecutionInfo

<!-- If there are risks that the release engineer should know about
document them here.
For example:
- Has an API/IDL been modified? Is it backwards/forwards compatible? If
not, what are the repecussions?
- Has a schema change been introduced? Is it possible to roll back?
- Has a feature flag been re-used for a new purpose? 
- Is there a potential performance concern? Is the change modifying core
task processing logic?
- If truly N/A, you can mark it as such -->
**Potential risks**
We are changing how we read data from Cassandra and if we are doing
incorrectly that could cause workflows to be corrupt/stuck.

We are also changing the argument to the parsing function and we had to
modify/use the whole result instead of passing only the "execution". If
this is wrong it could cause issues in parsing and affect workflows.

<!-- If this PR completes a user facing feature or changes functionality
add release notes here.
Your release notes should allow a user and the release engineer to
understand the changes with little context.
Always ensure that the description contains a link to the relevant
GitHub issue. -->
**Release notes**
Improve workflow corruption detection in Cassandra by reading from
next_event_id denormalized column and checking against checksum in
checksum verification.

<!-- Consider whether this change requires documentation updates in the
Cadence-Docs repo
- If yes: mention what needs updating (or link to docs PR in
cadence-docs repo)
- If in doubt, add a note about potential doc needs
- Only mark N/A if you're certain no docs are affected -->
**Documentation Changes**


---

## Reviewer Validation

**PR Description Quality** (check these before reviewing code):

- [ ] **"What changed"** provides a clear 1-2 line summary
  - [ ] Project Issue is linked
- [ ] **"Why"** explains the full motivation with sufficient context
- [ ] **Testing is documented:**
- [ ] Unit test commands are included (with exact `go test` invocation)
- [ ] Integration test setup/commands included (if integration tests
were run)
  - [ ] Canary testing details included (if canary was mentioned)
- [ ] **Potential risks** section is thoughtfully filled out (or
legitimately N/A)
- [ ] **Release notes** included if this completes a user-facing feature
- [ ] **Documentation** needs are addressed (or noted if uncertain)

---------

Signed-off-by: fimanishi <fimanishi@gmail.com>

Author: fimanishi

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

@@ -97,7 +97,10 @@ func (db *CDB) SelectCurrentWorkflow(
 	}
 
 	currentRunID := result["current_run_id"].(gocql.UUID).String()
-	executionInfo := parseWorkflowExecutionInfo(result["execution"].(map[string]interface{}))
+	executionInfo, err := parseWorkflowExecutionInfo(result)
+	if err != nil {
+		return nil, err
+	}
 	lastWriteVersion := constants.EmptyVersion
 	if result["workflow_last_write_version"] != nil {
 		lastWriteVersion = result["workflow_last_write_version"].(int64)
@@ -205,7 +208,10 @@ func (db *CDB) SelectWorkflowExecution(ctx context.Context, shardID int, domainI
 	}
 
 	state := &nosqlplugin.WorkflowExecution{}
-	info := parseWorkflowExecutionInfo(result["execution"].(map[string]interface{}))
+	info, err := parseWorkflowExecutionInfo(result)
+	if err != nil {
+		return nil, err
+	}
 	state.ExecutionInfo = info
 	state.VersionHistories = persistence.NewDataBlob(result["version_histories"].([]byte), constants.EncodingType(result["version_histories_encoding"].(string)))
 	// TODO: remove this after all 2DC workflows complete
@@ -357,8 +363,12 @@ func (db *CDB) SelectAllWorkflowExecutions(ctx context.Context, shardID int, pag
 			result = make(map[string]interface{})
 			continue
 		}
+		executionInfo, err := parseWorkflowExecutionInfo(result)
+		if err != nil {
+			return nil, nil, err
+		}
 		executions = append(executions, &persistence.InternalListConcreteExecutionsEntity{
-			ExecutionInfo:    parseWorkflowExecutionInfo(result["execution"].(map[string]interface{})),
+			ExecutionInfo:    executionInfo,
 			VersionHistories: persistence.NewDataBlob(result["version_histories"].([]byte), constants.EncodingType(result["version_histories_encoding"].(string))),
 		})
 		result = make(map[string]interface{})

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

@@ -312,7 +312,8 @@ const (
 	// TODO: remove replication_state after all 2DC workflows complete
 	templateGetWorkflowExecutionQuery = `SELECT execution, replication_state, activity_map, timer_map, ` +
 		`child_executions_map, request_cancel_map, signal_map, signal_requested, buffered_events_list, ` +
-		`buffered_replication_tasks_map, version_histories, version_histories_encoding, checksum ` +
+		`buffered_replication_tasks_map, version_histories, version_histories_encoding, checksum, ` +
+		`next_event_id ` +
 		`FROM executions ` +
 		`WHERE shard_id = ? ` +
 		`and type = ? ` +

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

@@ -22,6 +22,7 @@
 package cassandra
 
 import (
+	"fmt"
 	"time"
 
 	cql "github.com/gocql/gocql"
@@ -37,7 +38,11 @@ import (
 
 var _emptyUUID = cql.UUID{}
 
-func parseWorkflowExecutionInfo(result map[string]interface{}) *persistence.InternalWorkflowExecutionInfo {
+func parseWorkflowExecutionInfo(result map[string]interface{}) (*persistence.InternalWorkflowExecutionInfo, error) {
+	executionBlob, ok := result["execution"].(map[string]interface{})
+	if !ok {
+		return nil, fmt.Errorf("invalid execution blob format: missing or invalid 'execution' field")
+	}
 	info := &persistence.InternalWorkflowExecutionInfo{}
 	var completionEventData []byte
 	var completionEventEncoding constants.EncodingType
@@ -46,7 +51,7 @@ func parseWorkflowExecutionInfo(result map[string]interface{}) *persistence.Inte
 	var activeClusterSelectionPolicy []byte
 	var activeClusterSelectionPolicyEncoding constants.EncodingType
 
-	for k, v := range result {
+	for k, v := range executionBlob {
 		switch k {
 		case "domain_id":
 			info.DomainID = v.(gocql.UUID).String()
@@ -106,8 +111,6 @@ func parseWorkflowExecutionInfo(result map[string]interface{}) *persistence.Inte
 			info.LastFirstEventID = v.(int64)
 		case "last_event_task_id":
 			info.LastEventTaskID = v.(int64)
-		case "next_event_id":
-			info.NextEventID = v.(int64)
 		case "last_processed_event":
 			info.LastProcessedEvent = v.(int64)
 		case "start_time":
@@ -191,7 +194,12 @@ func parseWorkflowExecutionInfo(result map[string]interface{}) *persistence.Inte
 	info.CompletionEvent = persistence.NewDataBlob(completionEventData, completionEventEncoding)
 	info.AutoResetPoints = persistence.NewDataBlob(autoResetPoints, autoResetPointsEncoding)
 	info.ActiveClusterSelectionPolicy = persistence.NewDataBlob(activeClusterSelectionPolicy, activeClusterSelectionPolicyEncoding)
-	return info
+
+	if nextEventID, ok := result["next_event_id"].(int64); ok {
+		info.NextEventID = nextEventID
+	}
+
+	return info, nil
 }
 
 // TODO: remove this after all 2DC workflows complete