test: add SendToChannel helper and replace bare channel sends in functional tests (#9876)

## What changed?
Added a `SendToChannel(ctx, ch)` helper to `FunctionalTestBase`,
mirroring the existing `WaitForChannel`, and replaced all bare `ch <-
struct{}{}` sends in functional test bodies with calls to it.

## Why?
Bare channel sends in test bodies silently hang when the receiving
goroutine has stalled or died, causing tests to block until the overall
test timeout rather than failing fast with a useful message.
`WaitForChannel` already existed to solve this for receives,
`SendToChannel` closes the gap for sends.

## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)

## Potential risks
None, this is a pure test-helper addition.

Author: spkane31

tests/activity_api_pause_test.go

@@ -115,7 +115,7 @@ func (s *ActivityAPIPauseClientTestSuite) TestActivityPauseApi_WhileRunning() {
 	}, 5*time.Second, 500*time.Millisecond)
 
 	// unblock the activity
-	activityPausedCn <- struct{}{}
+	env.SendToChannel(ctx, activityPausedCn)
 	// make sure activity is paused on server and completed on the worker
 	s.EventuallyWithT(func(t *assert.CollectT) {
 		description, err := env.SdkClient().DescribeWorkflowExecution(ctx, workflowRun.GetID(), workflowRun.GetRunID())
@@ -264,7 +264,7 @@ func (s *ActivityAPIPauseClientTestSuite) TestActivityPauseApi_IncreaseAttemptsO
 	}, 5*time.Second, 500*time.Millisecond)
 
 	// End the activity
-	activityPausedCn <- struct{}{}
+	env.SendToChannel(ctx, activityPausedCn)
 
 	s.EventuallyWithT(func(t *assert.CollectT) {
 		description, err := env.SdkClient().DescribeWorkflowExecution(ctx, workflowRun.GetID(), workflowRun.GetRunID())
@@ -632,7 +632,7 @@ func (s *ActivityAPIPauseClientTestSuite) TestActivityPauseApi_WithReset() {
 	}, 5*time.Second, 100*time.Millisecond)
 
 	// let activity finish
-	activityCompleteCn <- struct{}{}
+	env.SendToChannel(ctx, activityCompleteCn)
 
 	// wait for workflow to finish
 	var out string

tests/nexus_workflow_test.go

@@ -1765,7 +1765,7 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationCancelBeforeStarted_Cancelati
 	})
 	s.NoError(err)
 
-	canStartCh <- struct{}{}
+	env.SendToChannel(ctx, canStartCh)
 	env.WaitForChannel(ctx, cancelSentCh)
 
 	// Terminate the workflow for good measure.

tests/pause_workflow_execution_test.go

@@ -233,7 +233,7 @@ func (s *PauseWorkflowExecutionSuite) TestPauseUnpauseWorkflowExecution() {
 	}, 5*time.Second, 200*time.Millisecond)
 
 	// Unblock the activity to complete the workflow.
-	s.activityCompletedCh <- struct{}{}
+	s.SendToChannel(ctx, s.activityCompletedCh)
 
 	// assert that the workflow completes now.
 	s.EventuallyWithT(func(t *assert.CollectT) {
@@ -591,7 +591,7 @@ func (s *PauseWorkflowExecutionSuite) TestQueryWorkflowWhenPaused() {
 	s.NotNil(unpauseResp)
 
 	// Unblock the activity and send the signal to complete the workflow.
-	s.activityCompletedCh <- struct{}{}
+	s.SendToChannel(ctx, s.activityCompletedCh)
 	err = s.SdkClient().SignalWorkflow(ctx, workflowID, runID, s.testEndSignal, "test end signal")
 	s.NoError(err)
 
@@ -818,7 +818,7 @@ func (s *PauseWorkflowExecutionSuite) TestPauseWorkflowExecutionAlreadyPaused()
 	}, 5*time.Second, 200*time.Millisecond)
 
 	// Unblock the activity and send the signal to complete the workflow.
-	s.activityCompletedCh <- struct{}{}
+	s.SendToChannel(ctx, s.activityCompletedCh)
 	err = s.SdkClient().SignalWorkflow(ctx, workflowID, runID, s.testEndSignal, "test end signal")
 	s.NoError(err)
 

tests/testcore/functional_test_base.go

@@ -659,6 +659,15 @@ func (s *FunctionalTestBase) WaitForChannel(ctx context.Context, ch chan struct{
 	}
 }
 
+func (s *FunctionalTestBase) SendToChannel(ctx context.Context, ch chan struct{}) {
+	s.T().Helper()
+	select {
+	case ch <- struct{}{}:
+	case <-ctx.Done():
+		s.FailNow("context timeout while sending to channel")
+	}
+}
+
 // TODO (alex): change to nsName namespace.Name
 func (s *FunctionalTestBase) SendSignal(nsName string, execution *commonpb.WorkflowExecution, signalName string,
 	input *commonpb.Payloads, identity string) error {

tests/versioning_test.go

@@ -2070,34 +2070,34 @@ func (s *VersioningIntegSuite) TestDispatchActivityUpgrade() {
 	s.WaitForChannel(ctx, startedWf)
 	rule2 := s.addRedirectRule(ctx, tq, v1, v11)
 	s.waitForRedirectRulePropagation(ctx, tq, rule2)
-	proceedWf <- struct{}{}
+	s.SendToChannel(ctx, proceedWf)
 
 	s.WaitForChannel(ctx, started11)
 	// wf assigned build ID should be updated by activity redirect
 	s.validateWorkflowBuildIds(ctx, run.GetID(), run.GetRunID(), v11, true, v1, "", []string{v1})
 	// let activity finish
-	proceed11 <- struct{}{}
+	s.SendToChannel(ctx, proceed11)
 
 	// wf replays on 1.1 so need to unblock it an extra time
 	s.WaitForChannel(ctx, startedWf)
-	proceedWf <- struct{}{}
+	s.SendToChannel(ctx, proceedWf)
 
 	s.WaitForChannel(ctx, startedWf)
 	rule2 = s.addRedirectRule(ctx, tq, v11, v12)
 	s.waitForRedirectRulePropagation(ctx, tq, rule2)
-	proceedWf <- struct{}{}
+	s.SendToChannel(ctx, proceedWf)
 
 	s.WaitForChannel(ctx, started12)
 	// wf assigned build ID should not be updated by independent activity redirect
 	s.validateWorkflowBuildIds(ctx, run.GetID(), run.GetRunID(), v11, true, v11, "", []string{v1})
 	// let activity finish
-	proceed12 <- struct{}{}
+	s.SendToChannel(ctx, proceed12)
 
 	// wf replays on 1.2 so need to unblock it two extra times
 	s.WaitForChannel(ctx, startedWf)
-	proceedWf <- struct{}{}
+	s.SendToChannel(ctx, proceedWf)
 	s.WaitForChannel(ctx, startedWf)
-	proceedWf <- struct{}{}
+	s.SendToChannel(ctx, proceedWf)
 
 	var out string
 	s.NoError(run.Get(ctx, &out))