potential solution to multiple async updates

temporalio · Nov 4, 2024 · 16b9ce3 · 16b9ce3
1 parent fd52f49
commit 16b9ce3
Show file tree

Hide file tree

Showing 3 changed files with 28 additions and 8 deletions.
diff --git a/internal/internal_workflow_test.go b/internal/internal_workflow_test.go
@@ -1407,6 +1407,18 @@ type updateCallback struct {
 	reject   func(error)
 	complete func(interface{}, error)
 	env      *TestWorkflowEnvironment
+	updateID string
+}
+
+// env and updateID are needed to cache update results for deduping purposes
+func newUpdateCallback(env *TestWorkflowEnvironment, updateID string, accept func(), reject func(error), complete func(interface{}, error)) *updateCallback {
+	return &updateCallback{
+		accept:   accept,
+		reject:   reject,
+		complete: complete,
+		env:      env,
+		updateID: updateID,
+	}
 }
 
 func (uc *updateCallback) Accept() {

diff --git a/internal/internal_workflow_testsuite.go b/internal/internal_workflow_testsuite.go
@@ -2750,6 +2750,8 @@ func (env *testWorkflowEnvironmentImpl) updateWorkflow(name string, id string, u
 			uc.Complete(env.updateMap[id].success, env.updateMap[id].err)
 		}, false)
 	} else {
+		// TODO: This doesn't account for multiple async updates
+		// would a UC -> ID map work? Would I have to use pointers?
 		env.currentUpdateId = id
 		env.postCallback(func() {
 			// Do not send any headers on test invocations
@@ -2768,6 +2770,7 @@ func (env *testWorkflowEnvironmentImpl) updateWorkflowByID(workflowID, name, id
 		if err != nil {
 			panic(err)
 		}
+		// TODO: handle dedup
 		workflowHandle.env.postCallback(func() {
 			workflowHandle.env.updateHandler(name, id, data, nil, uc)
 		}, true)

diff --git a/internal/workflow_testsuite_test.go b/internal/workflow_testsuite_test.go
@@ -273,15 +273,17 @@ func TestWorkflowIDUpdateWorkflowByID(t *testing.T) {
 	var suite WorkflowTestSuite
 	// Test UpdateWorkflowByID works with custom ID
 	env := suite.NewTestWorkflowEnvironment()
+	updateID := "id"
 	env.RegisterDelayedCallback(func() {
-		err := env.UpdateWorkflowByID("my-workflow-id", "update", "id", &updateCallback{
-			reject: func(err error) {
+		err := env.UpdateWorkflowByID("my-workflow-id", "update", updateID, newUpdateCallback(
+			env,
+			updateID,
+			func() {},
+			func(err error) {
 				require.Fail(t, "update should not be rejected")
 			},
-			accept:   func() {},
-			complete: func(interface{}, error) {},
-			env:      env,
-		}, "input")
+			func(interface{}, error) {},
+		), "input")
 		require.NoError(t, err)
 	}, time.Second)
 
@@ -500,7 +502,7 @@ func TestWorkflowUpdateOrderAcceptReject(t *testing.T) {
 
 func TestWorkflowDuplicateIDDedup(t *testing.T) {
 	var suite WorkflowTestSuite
-	// Test dev server rejects UpdateWorkflow with same ID
+	// Test dev server dedups UpdateWorkflow with same ID
 	env := suite.NewTestWorkflowEnvironment()
 	env.RegisterDelayedCallback(func() {
 		env.UpdateWorkflow("update", "id", &updateCallback{
@@ -517,7 +519,8 @@ func TestWorkflowDuplicateIDDedup(t *testing.T) {
 					require.Equal(t, 0, intResult)
 				}
 			},
-			env: env,
+			env:      env,
+			updateID: "id",
 		}, 0)
 	}, 0)
 
@@ -553,6 +556,7 @@ func TestWorkflowDuplicateIDDedup(t *testing.T) {
 		return Sleep(ctx, time.Hour)
 	})
 	require.NoError(t, env.GetWorkflowError())
+	require.True(t, false)
 }
 
 func TestAllHandlersFinished(t *testing.T) {
@@ -802,6 +806,7 @@ func TestWorkflowUpdateLogger(t *testing.T) {
 			},
 			accept:   func() {},
 			complete: func(interface{}, error) {},
+			env:      env,
 		})
 	}, 0)