Use safeAsTime and safeAsDuration

internal/activity.go

@@ -286,11 +286,11 @@ func WithActivityTask(
 	interceptors []WorkerInterceptor,
 	client *WorkflowClient,
 ) (context.Context, error) {
-	scheduled := task.GetScheduledTime().AsTime()
-	started := task.GetStartedTime().AsTime()
-	scheduleToCloseTimeout := task.GetScheduleToCloseTimeout().AsDuration()
-	startToCloseTimeout := task.GetStartToCloseTimeout().AsDuration()
-	heartbeatTimeout := task.GetHeartbeatTimeout().AsDuration()
+	scheduled := safeAsTime(task.GetScheduledTime())
+	started := safeAsTime(task.GetStartedTime())
+	scheduleToCloseTimeout := safeAsDuration(task.GetScheduleToCloseTimeout())
+	startToCloseTimeout := safeAsDuration(task.GetStartToCloseTimeout())
+	heartbeatTimeout := safeAsDuration(task.GetHeartbeatTimeout())
 	deadline := calculateActivityDeadline(scheduled, started, scheduleToCloseTimeout, startToCloseTimeout)
 
 	logger = log.With(logger,

internal/internal_deployment_client.go

@@ -73,7 +73,7 @@ func deploymentToProto(deploymentID Deployment) *deployment.Deployment {
 func deploymentListEntryFromProto(deployment *deployment.DeploymentListInfo) *DeploymentListEntry {
 	return &DeploymentListEntry{
 		Deployment: deploymentFromProto(deployment.GetDeployment()),
-		CreateTime: deployment.GetCreateTime().AsTime(),
+		CreateTime: safeAsTime(deployment.GetCreateTime()),
 		IsCurrent:  deployment.GetIsCurrent(),
 	}
 }
@@ -84,7 +84,7 @@ func deploymentTaskQueuesInfoFromProto(tqsInfo []*deployment.DeploymentInfo_Task
 		result = append(result, DeploymentTaskQueueInfo{
 			Name:            info.GetName(),
 			Type:            TaskQueueType(info.GetType()),
-			FirstPollerTime: info.GetFirstPollerTime().AsTime(),
+			FirstPollerTime: safeAsTime(info.GetFirstPollerTime()),
 		})
 	}
 	return result
@@ -93,7 +93,7 @@ func deploymentTaskQueuesInfoFromProto(tqsInfo []*deployment.DeploymentInfo_Task
 func deploymentInfoFromProto(deploymentInfo *deployment.DeploymentInfo) DeploymentInfo {
 	return DeploymentInfo{
 		Deployment:      deploymentFromProto(deploymentInfo.GetDeployment()),
-		CreateTime:      deploymentInfo.GetCreateTime().AsTime(),
+		CreateTime:      safeAsTime(deploymentInfo.GetCreateTime()),
 		IsCurrent:       deploymentInfo.GetIsCurrent(),
 		TaskQueuesInfos: deploymentTaskQueuesInfoFromProto(deploymentInfo.GetTaskQueueInfos()),
 		Metadata:        deploymentInfo.GetMetadata(),
@@ -110,7 +110,7 @@ func deploymentReachabilityInfoFromProto(response *workflowservice.GetDeployment
 	return DeploymentReachabilityInfo{
 		DeploymentInfo: deploymentInfoFromProto(response.GetDeploymentInfo()),
 		Reachability:   DeploymentReachability(response.GetReachability()),
-		LastUpdateTime: response.GetLastUpdateTime().AsTime(),
+		LastUpdateTime: safeAsTime(response.GetLastUpdateTime()),
 	}
 }
 

internal/internal_event_handlers.go

@@ -1194,7 +1194,7 @@ func (weh *workflowExecutionEventHandlerImpl) ProcessEvent(
 		// No Operation
 	case enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED:
 		// Set replay clock.
-		weh.SetCurrentReplayTime(event.GetEventTime().AsTime())
+		weh.SetCurrentReplayTime(safeAsTime(event.GetEventTime()))
 		// Update workflow info fields
 		weh.workflowInfo.currentHistoryLength = int(event.EventId)
 		weh.workflowInfo.continueAsNewSuggested = event.GetWorkflowTaskStartedEventAttributes().GetSuggestContinueAsNew()

internal/internal_nexus_task_poller.go

@@ -123,7 +123,7 @@ func (ntp *nexusTaskPoller) ProcessTask(task interface{}) error {
 	// Schedule-to-start (from the time the request hit the frontend).
 	// Note that this metric does not include the service and operation name as they are not relevant when polling from
 	// the Nexus task queue.
-	scheduleToStartLatency := executionStartTime.Sub(response.GetRequest().GetScheduledTime().AsTime())
+	scheduleToStartLatency := executionStartTime.Sub(safeAsTime(response.GetRequest().GetScheduledTime()))
 	ntp.metricsHandler.WithTags(metrics.TaskQueueTags(ntp.taskQueueName)).Timer(metrics.NexusTaskScheduleToStartLatency).Record(scheduleToStartLatency)
 
 	nctx, handlerErr := ntp.taskHandler.newNexusOperationContext(response)
@@ -184,7 +184,7 @@ func (ntp *nexusTaskPoller) ProcessTask(task interface{}) error {
 	// E2E latency, from frontend until we finished reporting completion.
 	nctx.metricsHandler.
 		Timer(metrics.NexusTaskEndToEndLatency).
-		Record(time.Since(response.GetRequest().GetScheduledTime().AsTime()))
+		Record(time.Since(safeAsTime(response.GetRequest().GetScheduledTime())))
 	return nil
 }
 

internal/internal_schedule_client.go

@@ -419,21 +419,21 @@ func convertFromPBScheduleSpec(scheduleSpec *schedulepb.ScheduleSpec) *ScheduleS
 	intervals := make([]ScheduleIntervalSpec, len(scheduleSpec.GetInterval()))
 	for i, s := range scheduleSpec.GetInterval() {
 		intervals[i] = ScheduleIntervalSpec{
-			Every:  s.Interval.AsDuration(),
-			Offset: s.Phase.AsDuration(),
+			Every:  safeAsDuration(s.Interval),
+			Offset: safeAsDuration(s.Phase),
 		}
 	}
 
 	skip := convertFromPBScheduleCalendarSpecList(scheduleSpec.GetExcludeStructuredCalendar())
 
 	startAt := time.Time{}
 	if scheduleSpec.GetStartTime() != nil {
-		startAt = scheduleSpec.GetStartTime().AsTime()
+		startAt = safeAsTime(scheduleSpec.GetStartTime())
 	}
 
 	endAt := time.Time{}
 	if scheduleSpec.GetEndTime() != nil {
-		endAt = scheduleSpec.GetEndTime().AsTime()
+		endAt = safeAsTime(scheduleSpec.GetEndTime())
 	}
 
 	return &ScheduleSpec{
@@ -442,7 +442,7 @@ func convertFromPBScheduleSpec(scheduleSpec *schedulepb.ScheduleSpec) *ScheduleS
 		Skip:         skip,
 		StartAt:      startAt,
 		EndAt:        endAt,
-		Jitter:       scheduleSpec.GetJitter().AsDuration(),
+		Jitter:       safeAsDuration(scheduleSpec.GetJitter()),
 		TimeZoneName: scheduleSpec.GetTimezoneName(),
 	}
 }
@@ -468,7 +468,7 @@ func scheduleDescriptionFromPB(
 
 	nextActionTimes := make([]time.Time, len(describeResponse.Info.GetFutureActionTimes()))
 	for i, t := range describeResponse.Info.GetFutureActionTimes() {
-		nextActionTimes[i] = t.AsTime()
+		nextActionTimes[i] = safeAsTime(t)
 	}
 
 	actionDescription, err := convertFromPBScheduleAction(logger, dc, describeResponse.Schedule.Action)
@@ -488,7 +488,7 @@ func scheduleDescriptionFromPB(
 			Spec:   convertFromPBScheduleSpec(describeResponse.Schedule.Spec),
 			Policy: &SchedulePolicies{
 				Overlap:        describeResponse.Schedule.Policies.GetOverlapPolicy(),
-				CatchupWindow:  describeResponse.Schedule.Policies.GetCatchupWindow().AsDuration(),
+				CatchupWindow:  safeAsDuration(describeResponse.Schedule.Policies.GetCatchupWindow()),
 				PauseOnFailure: describeResponse.Schedule.Policies.GetPauseOnFailure(),
 			},
 			State: &ScheduleState{
@@ -505,8 +505,8 @@ func scheduleDescriptionFromPB(
 			RunningWorkflows:              runningWorkflows,
 			RecentActions:                 recentActions,
 			NextActionTimes:               nextActionTimes,
-			CreatedAt:                     describeResponse.Info.GetCreateTime().AsTime(),
-			LastUpdateAt:                  describeResponse.Info.GetUpdateTime().AsTime(),
+			CreatedAt:                     safeAsTime(describeResponse.Info.GetCreateTime()),
+			LastUpdateAt:                  safeAsTime(describeResponse.Info.GetUpdateTime()),
 		},
 		Memo:                  describeResponse.Memo,
 		SearchAttributes:      searchAttributes,
@@ -553,7 +553,7 @@ func convertFromPBScheduleListEntry(schedule *schedulepb.ScheduleListEntry) *Sch
 
 	nextActionTimes := make([]time.Time, len(schedule.Info.GetFutureActionTimes()))
 	for i, t := range schedule.Info.GetFutureActionTimes() {
-		nextActionTimes[i] = t.AsTime()
+		nextActionTimes[i] = safeAsTime(t)
 	}
 
 	return &ScheduleListEntry{
@@ -712,9 +712,9 @@ func convertFromPBScheduleAction(
 			Workflow:                 workflow.WorkflowType.GetName(),
 			Args:                     args,
 			TaskQueue:                workflow.TaskQueue.GetName(),
-			WorkflowExecutionTimeout: workflow.GetWorkflowExecutionTimeout().AsDuration(),
-			WorkflowRunTimeout:       workflow.GetWorkflowRunTimeout().AsDuration(),
-			WorkflowTaskTimeout:      workflow.GetWorkflowTaskTimeout().AsDuration(),
+			WorkflowExecutionTimeout: safeAsDuration(workflow.GetWorkflowExecutionTimeout()),
+			WorkflowRunTimeout:       safeAsDuration(workflow.GetWorkflowRunTimeout()),
+			WorkflowTaskTimeout:      safeAsDuration(workflow.GetWorkflowTaskTimeout()),
 			RetryPolicy:              convertFromPBRetryPolicy(workflow.RetryPolicy),
 			Memo:                     memos,
 			TypedSearchAttributes:    searchAttrs,
@@ -842,8 +842,8 @@ func convertFromPBScheduleActionResultList(aa []*schedulepb.ScheduleActionResult
 			}
 		}
 		recentActions[i] = ScheduleActionResult{
-			ScheduleTime:        a.GetScheduleTime().AsTime(),
-			ActualTime:          a.GetActualTime().AsTime(),
+			ScheduleTime:        safeAsTime(a.GetScheduleTime()),
+			ActualTime:          safeAsTime(a.GetActualTime()),
 			StartWorkflowResult: workflowExecution,
 		}
 	}

internal/internal_task_handlers.go

@@ -727,12 +727,12 @@ func (wth *workflowTaskHandlerImpl) createWorkflowContext(task *workflowservice.
 		FirstRunID:               attributes.FirstExecutionRunId,
 		WorkflowType:             WorkflowType{Name: task.WorkflowType.GetName()},
 		TaskQueueName:            taskQueue.GetName(),
-		WorkflowExecutionTimeout: attributes.GetWorkflowExecutionTimeout().AsDuration(),
-		WorkflowRunTimeout:       attributes.GetWorkflowRunTimeout().AsDuration(),
-		WorkflowTaskTimeout:      attributes.GetWorkflowTaskTimeout().AsDuration(),
+		WorkflowExecutionTimeout: safeAsDuration(attributes.GetWorkflowExecutionTimeout()),
+		WorkflowRunTimeout:       safeAsDuration(attributes.GetWorkflowRunTimeout()),
+		WorkflowTaskTimeout:      safeAsDuration(attributes.GetWorkflowTaskTimeout()),
 		Namespace:                wth.namespace,
 		Attempt:                  attributes.GetAttempt(),
-		WorkflowStartTime:        startedEvent.GetEventTime().AsTime(),
+		WorkflowStartTime:        safeAsTime(startedEvent.GetEventTime()),
 		lastCompletionResult:     attributes.LastCompletionResult,
 		lastFailure:              attributes.ContinuedFailure,
 		CronSchedule:             attributes.CronSchedule,
@@ -2211,7 +2211,7 @@ func (ath *activityTaskHandlerImpl) Execute(taskQueue string, t *workflowservice
 	canCtx, cancel := context.WithCancelCause(rootCtx)
 	defer cancel(nil)
 
-	heartbeatThrottleInterval := ath.getHeartbeatThrottleInterval(t.GetHeartbeatTimeout().AsDuration())
+	heartbeatThrottleInterval := ath.getHeartbeatThrottleInterval(safeAsDuration(t.GetHeartbeatTimeout()))
 	invoker := newServiceInvoker(
 		t.TaskToken, ath.identity, ath.client.workflowService, ath.metricsHandler, cancel, heartbeatThrottleInterval,
 		ath.workerStopCh, ath.namespace)

internal/internal_task_pollers.go

@@ -864,7 +864,7 @@ func (wtp *workflowTaskPoller) poll(ctx context.Context) (taskForWorker, error)
 	metricsHandler := wtp.metricsHandler.WithTags(metrics.WorkflowTags(response.WorkflowType.GetName()))
 	metricsHandler.Counter(metrics.WorkflowTaskQueuePollSucceedCounter).Inc(1)
 
-	scheduleToStartLatency := response.GetStartedTime().AsTime().Sub(response.GetScheduledTime().AsTime())
+	scheduleToStartLatency := safeAsTime(response.GetStartedTime()).Sub(safeAsTime(response.GetScheduledTime()))
 	metricsHandler.Timer(metrics.WorkflowTaskScheduleToStartLatency).Record(scheduleToStartLatency)
 	return task, nil
 }
@@ -1032,7 +1032,7 @@ func (atp *activityTaskPoller) poll(ctx context.Context) (taskForWorker, error)
 	activityType := response.ActivityType.GetName()
 	metricsHandler := atp.metricsHandler.WithTags(metrics.ActivityTags(workflowType, activityType, atp.taskQueueName))
 
-	scheduleToStartLatency := response.GetStartedTime().AsTime().Sub(response.GetCurrentAttemptScheduledTime().AsTime())
+	scheduleToStartLatency := safeAsTime(response.GetStartedTime()).Sub(safeAsTime(response.GetCurrentAttemptScheduledTime()))
 	metricsHandler.Timer(metrics.ActivityScheduleToStartLatency).Record(scheduleToStartLatency)
 
 	return &activityTask{task: response}, nil
@@ -1103,7 +1103,7 @@ func (atp *activityTaskPoller) ProcessTask(task interface{}) error {
 	if _, ok := request.(*workflowservice.RespondActivityTaskCompletedRequest); ok {
 		activityMetricsHandler.
 			Timer(metrics.ActivitySucceedEndToEndLatency).
-			Record(time.Since(activityTask.task.GetScheduledTime().AsTime()))
+			Record(time.Since(safeAsTime(activityTask.task.GetScheduledTime())))
 	}
 	return nil
 }

internal/internal_versioning_client.go

@@ -352,7 +352,7 @@ func pollerInfoFromResponse(response *taskqueuepb.PollerInfo) TaskQueuePollerInf
 
 	lastAccessTime := time.Time{}
 	if response.GetLastAccessTime() != nil {
-		lastAccessTime = response.GetLastAccessTime().AsTime()
+		lastAccessTime = safeAsTime(response.GetLastAccessTime())
 	}
 
 	return TaskQueuePollerInfo{
@@ -388,7 +388,7 @@ func statsFromResponse(stats *taskqueuepb.TaskQueueStats) *TaskQueueStats {
 
 	return &TaskQueueStats{
 		ApproximateBacklogCount: stats.GetApproximateBacklogCount(),
-		ApproximateBacklogAge:   stats.GetApproximateBacklogAge().AsDuration(),
+		ApproximateBacklogAge:   safeAsDuration(stats.GetApproximateBacklogAge()),
 		TasksAddRate:            stats.TasksAddRate,
 		TasksDispatchRate:       stats.TasksDispatchRate,
 		BacklogIncreaseRate:     stats.TasksAddRate - stats.TasksDispatchRate,
@@ -429,7 +429,7 @@ func taskQueueVersioningInfoFromResponse(info *taskqueuepb.TaskQueueVersioningIn
 		CurrentVersion:           info.CurrentVersion,
 		RampingVersion:           info.RampingVersion,
 		RampingVersionPercentage: info.RampingVersionPercentage,
-		UpdateTime:               info.UpdateTime.AsTime(),
+		UpdateTime:               safeAsTime(info.UpdateTime),
 	}
 }
 

internal/internal_versioning_client_test.go

@@ -53,7 +53,7 @@ func Test_DetectEnhancedNotSupported_fromProtoResponse(t *testing.T) {
 
 func Test_TaskQueueDescription_fromProtoResponse(t *testing.T) {
 	nowProto := timestamppb.Now()
-	now := nowProto.AsTime()
+	now := safeAsTime(nowProto)
 	tests := []struct {
 		name     string
 		response *workflowservice.DescribeTaskQueueResponse

internal/internal_worker_deployment_client.go

@@ -11,6 +11,7 @@ import (
 	"go.temporal.io/api/deployment/v1"
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/sdk/converter"
+	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -29,6 +30,15 @@ func safeAsTime(timestamp *timestamppb.Timestamp) time.Time {
 	}
 }
 
+// safeAsDuration ensures that a nil proto duration makes `AsDuration()` return 0.
+func safeAsDuration(duration *durationpb.Duration) time.Duration {
+	if duration == nil {
+		return time.Duration(0)
+	} else {
+		return duration.AsDuration()
+	}
+}
+
 type (
 	// WorkerDeploymentClient is the client for managing worker deployments.
 	workerDeploymentClient struct {

internal/internal_workflow_client.go

@@ -1597,10 +1597,10 @@ func (workflowRun *workflowRunImpl) GetWithOptions(
 			TaskQueueName: attributes.GetTaskQueue().GetName(),
 		}
 		if attributes.WorkflowRunTimeout != nil {
-			err.WorkflowRunTimeout = attributes.WorkflowRunTimeout.AsDuration()
+			err.WorkflowRunTimeout = safeAsDuration(attributes.WorkflowRunTimeout)
 		}
 		if attributes.WorkflowTaskTimeout != nil {
-			err.WorkflowTaskTimeout = attributes.WorkflowTaskTimeout.AsDuration()
+			err.WorkflowTaskTimeout = safeAsDuration(attributes.WorkflowTaskTimeout)
 		}
 		return err
 	default:
@@ -2223,12 +2223,12 @@ func (w *workflowClientInterceptor) DescribeWorkflow(
 
 	var closeTime *time.Time
 	if info.GetCloseTime().IsValid() {
-		t := info.GetCloseTime().AsTime()
+		t := safeAsTime(info.GetCloseTime())
 		closeTime = &t
 	}
 	var executionTime *time.Time
 	if info.GetExecutionTime().IsValid() {
-		t := info.GetExecutionTime().AsTime()
+		t := safeAsTime(info.GetExecutionTime())
 		executionTime = &t
 	}
 
@@ -2262,7 +2262,7 @@ func (w *workflowClientInterceptor) DescribeWorkflow(
 		Status:                  info.GetStatus(),
 		ParentWorkflowExecution: parentWorkflowExecution,
 		RootWorkflowExecution:   rootWorkflowExecution,
-		WorkflowStartTime:       info.GetStartTime().AsTime(),
+		WorkflowStartTime:       safeAsTime(info.GetStartTime()),
 		ExecutionTime:           executionTime,
 		WorkflowCloseTime:       closeTime,
 		HistoryLength:           int(info.GetHistoryLength()),