Work on inverted workers

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/stretchr/testify v1.8.3
 	go.temporal.io/api v1.21.0
 	go.uber.org/atomic v1.9.0
+	golang.org/x/sys v0.8.0
 	golang.org/x/time v0.3.0
 	google.golang.org/grpc v1.55.0
 	google.golang.org/protobuf v1.30.0

internal/activity.go

@@ -260,7 +260,7 @@ func WithActivityTask(
 	scheduleToCloseTimeout := common.DurationValue(task.GetScheduleToCloseTimeout())
 	startToCloseTimeout := common.DurationValue(task.GetStartToCloseTimeout())
 	heartbeatTimeout := common.DurationValue(task.GetHeartbeatTimeout())
-	deadline := calculateActivityDeadline(scheduled, started, scheduleToCloseTimeout, startToCloseTimeout)
+	deadline := CalculateActivityDeadline(scheduled, started, scheduleToCloseTimeout, startToCloseTimeout)
 
 	logger = log.With(logger,
 		tagActivityID, task.ActivityId,
@@ -331,7 +331,7 @@ func WithLocalActivityTask(
 	if scheduleToCloseTimeout == 0 {
 		scheduleToCloseTimeout = startToCloseTimeout
 	}
-	deadline := calculateActivityDeadline(task.scheduledTime, startedTime, scheduleToCloseTimeout, startToCloseTimeout)
+	deadline := CalculateActivityDeadline(task.scheduledTime, startedTime, scheduleToCloseTimeout, startToCloseTimeout)
 	if task.attempt > 1 && !task.expireTime.IsZero() && task.expireTime.Before(deadline) {
 		// this is attempt and expire time is before SCHEDULE_TO_CLOSE timeout
 		deadline = task.expireTime
@@ -382,7 +382,7 @@ func newActivityContext(
 	return ctx, nil
 }
 
-func calculateActivityDeadline(scheduled, started time.Time, scheduleToCloseTimeout, startToCloseTimeout time.Duration) time.Time {
+func CalculateActivityDeadline(scheduled, started time.Time, scheduleToCloseTimeout, startToCloseTimeout time.Duration) time.Time {
 	startToCloseDeadline := started.Add(startToCloseTimeout)
 	if scheduleToCloseTimeout > 0 {
 		scheduleToCloseDeadline := scheduled.Add(scheduleToCloseTimeout)

internal/client.go

@@ -460,6 +460,8 @@ type (
 		// worker options, the ones here wrap the ones in worker options. The same
 		// interceptor should not be set here and in worker options.
 		Interceptors []ClientInterceptor
+
+		overrideWorkflowService workflowservice.WorkflowServiceClient
 	}
 
 	// HeadersProvider returns a map of gRPC headers that should be used on every request.
@@ -669,6 +671,10 @@ type (
 	}
 )
 
+func OverrideWorkflowService(c *ClientOptions, workflowService workflowservice.WorkflowServiceClient) {
+	c.overrideWorkflowService = workflowService
+}
+
 // DialClient creates a client and attempts to connect to the server.
 func DialClient(options ClientOptions) (Client, error) {
 	options.ConnectionOptions.disableEagerConnection = false
@@ -719,19 +725,23 @@ func newClient(options ClientOptions, existing *WorkflowClient) (Client, error)
 	}
 
 	// Dial or use existing connection
+	serviceClient := options.overrideWorkflowService
 	var connection *grpc.ClientConn
 	var err error
-	if existing == nil {
-		options.ConnectionOptions.excludeInternalFromRetry = uberatomic.NewBool(false)
-		connection, err = dial(newDialParameters(&options, options.ConnectionOptions.excludeInternalFromRetry))
-		if err != nil {
-			return nil, err
+	if serviceClient == nil {
+		if existing == nil {
+			options.ConnectionOptions.excludeInternalFromRetry = uberatomic.NewBool(false)
+			connection, err = dial(newDialParameters(&options, options.ConnectionOptions.excludeInternalFromRetry))
+			if err != nil {
+				return nil, err
+			}
+		} else {
+			connection = existing.conn
 		}
-	} else {
-		connection = existing.conn
+		serviceClient = workflowservice.NewWorkflowServiceClient(connection)
 	}
 
-	client := NewServiceClient(workflowservice.NewWorkflowServiceClient(connection), connection, options)
+	client := NewServiceClient(serviceClient, connection, options)
 
 	// If using existing connection, always load its capabilities and use them for
 	// the new connection. Otherwise, only load server capabilities eagerly if not

worker/invertedworker/durable_store.go

@@ -0,0 +1,14 @@
+package invertedworker
+
+import (
+	"context"
+
+	"go.temporal.io/api/common/v1"
+	"go.temporal.io/api/history/v1"
+)
+
+type DurableStore interface {
+	AppendPartialHistory(context.Context, *common.WorkflowExecution, *history.History) error
+	GetHistory(context.Context, *common.WorkflowExecution) (*history.History, error)
+	PurgeHistory(context.Context, *common.WorkflowExecution) error
+}