Fix regression in Go SDK where Batch Timer interval for heartbeats is overcalculated (#225)

As part of switching from variables representing time in seconds to time.Duration, we forgot to correctly update the line of code that calculates the interval for sending the next heartbeat. We were multiplying by an extra factor of one billion, which either caused an overflow (so that we were constantly sending heartbeats) or resulted in a large number (where we would never send a heartbeat after the first one).

Author: mastermanu

internal/internal_task_handlers.go

@@ -1625,7 +1625,7 @@ func (i *temporalInvoker) Heartbeat(details *commonpb.Payloads, skipBatching boo
 		}
 
 		// We set a deadline at 80% of the timeout.
-		duration := time.Duration(0.8*float32(deadlineToTrigger)) * time.Second
+		duration := time.Duration(0.8 * float64(deadlineToTrigger))
 		i.hbBatchEndTimer = time.NewTimer(duration)
 
 		go func() {

internal/internal_task_handlers_test.go

@@ -1306,19 +1306,44 @@ func (t *TaskHandlersTestSuite) TestLocalActivityRetry_WorkflowTaskHeartbeatFail
 func (t *TaskHandlersTestSuite) TestHeartBeat_NoError() {
 	mockCtrl := gomock.NewController(t.T())
 	mockService := workflowservicemock.NewMockWorkflowServiceClient(mockCtrl)
-
+	invocationChannel := make(chan int, 2)
 	heartbeatResponse := workflowservice.RecordActivityTaskHeartbeatResponse{CancelRequested: false}
-	mockService.EXPECT().RecordActivityTaskHeartbeat(gomock.Any(), gomock.Any(), gomock.Any()).Return(&heartbeatResponse, nil)
+	mockService.EXPECT().
+		RecordActivityTaskHeartbeat(gomock.Any(), gomock.Any(), gomock.Any()).
+		Do(func(_ interface{}, _ interface{}, _ ...interface{}) { invocationChannel <- 1 }).
+		Return(&heartbeatResponse, nil).
+		Times(2)
 
 	temporalInvoker := &temporalInvoker{
-		identity:  "Test_Temporal_Invoker",
-		service:   mockService,
-		taskToken: nil,
+		identity:         "Test_Temporal_Invoker",
+		service:          mockService,
+		taskToken:        nil,
+		heartBeatTimeout: time.Second,
 	}
 
 	heartbeatErr := temporalInvoker.Heartbeat(nil, false)
+	t.NoError(heartbeatErr)
+
+	select {
+	case <-invocationChannel:
+	case <-time.After(3 * time.Second):
+		t.Fail("did not get expected 1st call to record heartbeat")
+	}
 
+	heartbeatErr = temporalInvoker.Heartbeat(nil, false)
 	t.NoError(heartbeatErr)
+
+	select {
+	case <-invocationChannel:
+		t.Fail("got unexpected call to record heartbeat. 2nd call should come via batch timer")
+	default:
+	}
+
+	select {
+	case <-invocationChannel:
+	case <-time.After(3 * time.Second):
+		t.Fail("did not get expected 2nd call to record heartbeat via batch timer")
+	}
 }
 
 func (t *TaskHandlersTestSuite) TestHeartBeat_NilResponseWithError() {

test/activity_test.go

@@ -85,6 +85,17 @@ func (a *Activities) HeartbeatAndSleep(ctx context.Context, seq int, delay time.
 	return seq, nil
 }
 
+func (a *Activities) LongRunningHeartbeat(ctx context.Context, delay time.Duration, recordHeartbeatDelay time.Duration) error {
+	a.append("longRunningHeartbeat")
+	endTime := time.Now().Add(delay)
+	for time.Now().Before(endTime) {
+		activity.RecordHeartbeat(ctx)
+		time.Sleep(recordHeartbeatDelay)
+	}
+
+	return nil
+}
+
 func (a *Activities) fail(_ context.Context) error {
 	a.append("fail")
 	return errFailOnPurpose

test/integration_test.go

@@ -193,6 +193,13 @@ func (ts *IntegrationTestSuite) TestActivityRetryOnHBTimeout() {
 	ts.EqualValues(expected, ts.activities.invoked())
 }
 
+func (ts *IntegrationTestSuite) TestLongRunningActivityWithHB() {
+	var expected []string
+	err := ts.executeWorkflow("test-long-running-activity-with-hb", ts.workflows.LongRunningActivityWithHB, &expected)
+	ts.NoError(err)
+	ts.EqualValues(expected, ts.activities.invoked())
+}
+
 func (ts *IntegrationTestSuite) TestContinueAsNew() {
 	var result int
 	err := ts.executeWorkflow("test-continueasnew", ts.workflows.ContinueAsNew, &result, 4, ts.taskQueueName)

test/workflow_test.go

@@ -137,6 +137,24 @@ func (w *Workflows) ActivityRetryOnTimeout(ctx workflow.Context, timeoutType enu
 	return []string{"sleep", "sleep", "sleep"}, nil
 }
 
+func (w *Workflows) LongRunningActivityWithHB(ctx workflow.Context) ([]string, error) {
+	opts := w.defaultActivityOptionsWithRetry()
+	opts.HeartbeatTimeout = 3 * time.Second
+	opts.ScheduleToCloseTimeout = time.Second * 12
+	opts.StartToCloseTimeout = time.Second * 12
+	opts.RetryPolicy = &internal.RetryPolicy{
+		MaximumAttempts: 1,
+	}
+	ctx = workflow.WithActivityOptions(ctx, opts)
+
+	err := workflow.ExecuteActivity(ctx, "LongRunningHeartbeat", 8*time.Second, 300*time.Millisecond).Get(ctx, nil)
+	if err != nil {
+		return nil, fmt.Errorf("expected activity to succeed but it failed: %v", err)
+	}
+
+	return []string{"longRunningHeartbeat"}, nil
+}
+
 func (w *Workflows) ActivityRetryOnHBTimeout(ctx workflow.Context) ([]string, error) {
 	opts := w.defaultActivityOptionsWithRetry()
 	opts.HeartbeatTimeout = time.Second
@@ -718,6 +736,7 @@ func (w *Workflows) register(worker worker.Worker) {
 	worker.RegisterWorkflow(w.InspectActivityInfo)
 	worker.RegisterWorkflow(w.InspectLocalActivityInfo)
 	worker.RegisterWorkflow(w.LargeQueryResultWorkflow)
+	worker.RegisterWorkflow(w.LongRunningActivityWithHB)
 	worker.RegisterWorkflow(w.RetryTimeoutStableErrorWorkflow)
 	worker.RegisterWorkflow(w.SimplestWorkflow)
 	worker.RegisterWorkflow(w.WorkflowWithLocalActivityCtxPropagation)