Merge pull request #22 from JoshVanL/workflow-rerun

Adds RerunWorkflowFromEvento

Author: JoshVanL

api/orchestration.go

@@ -47,6 +47,10 @@ type OrchestrationIdReusePolicy = protos.OrchestrationIdReusePolicy
 // InstanceID is a unique identifier for an orchestration instance.
 type InstanceID string
 
+func (i InstanceID) String() string {
+	return string(i)
+}
+
 // NewOrchestrationOptions configures options for starting a new orchestration.
 type NewOrchestrationOptions func(*protos.CreateInstanceRequest) error
 
@@ -62,6 +66,8 @@ type TerminateOptions func(*protos.TerminateRequest) error
 // PurgeOptions is a set of options for purging an orchestration.
 type PurgeOptions func(*protos.PurgeInstancesRequest) error
 
+type RerunOptions func(*protos.RerunWorkflowFromEventRequest) error
+
 // WithInstanceID configures an explicit orchestration instance ID. If not specified,
 // a random UUID value will be used for the orchestration instance ID.
 func WithInstanceID(id InstanceID) NewOrchestrationOptions {
@@ -181,8 +187,34 @@ func OrchestrationMetadataIsRunning(o *protos.OrchestrationMetadata) bool {
 }
 
 func OrchestrationMetadataIsComplete(o *protos.OrchestrationMetadata) bool {
-	return o.RuntimeStatus == protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED ||
-		o.RuntimeStatus == protos.OrchestrationStatus_ORCHESTRATION_STATUS_FAILED ||
-		o.RuntimeStatus == protos.OrchestrationStatus_ORCHESTRATION_STATUS_TERMINATED ||
-		o.RuntimeStatus == protos.OrchestrationStatus_ORCHESTRATION_STATUS_CANCELED
+	return o.GetRuntimeStatus() == protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED ||
+		o.GetRuntimeStatus() == protos.OrchestrationStatus_ORCHESTRATION_STATUS_FAILED ||
+		o.GetRuntimeStatus() == protos.OrchestrationStatus_ORCHESTRATION_STATUS_TERMINATED ||
+		o.GetRuntimeStatus() == protos.OrchestrationStatus_ORCHESTRATION_STATUS_CANCELED
+}
+
+func WithRerunInput(input any) RerunOptions {
+	return func(req *protos.RerunWorkflowFromEventRequest) error {
+		req.OverwriteInput = true
+
+		if input == nil {
+			return nil
+		}
+
+		bytes, err := json.Marshal(input)
+		if err != nil {
+			return err
+		}
+
+		req.Input = wrapperspb.String(string(bytes))
+
+		return nil
+	}
+}
+
+func WithRerunNewInstanceID(id InstanceID) RerunOptions {
+	return func(req *protos.RerunWorkflowFromEventRequest) error {
+		req.NewInstanceID = id.String()
+		return nil
+	}
 }

api/protos/orchestrator_service.pb.go

@@ -32,6 +32,7 @@ type (
 	DurableTimer                     = protos.DurableTimer
 	OrchestrationRuntimeState        = protos.OrchestrationRuntimeState
 	OrchestrationRuntimeStateMessage = protos.OrchestrationRuntimeStateMessage
+	RerunWorkflowFromEventRequest    = protos.RerunWorkflowFromEventRequest
 )
 
 type OrchestrationIdReusePolicyOptions func(*protos.OrchestrationIdReusePolicy) error
@@ -68,6 +69,12 @@ type Backend interface {
 	// wraps a ExecutionStarted event.
 	CreateOrchestrationInstance(context.Context, *HistoryEvent, ...OrchestrationIdReusePolicyOptions) error
 
+	// RerunWorkflowFromEvent reruns a workflow from a specific event ID of some
+	// source instance ID. If not given, a random new instance ID will be
+	// generated and returned. Can optionally give a new input to the target
+	// event ID to rerun from.
+	RerunWorkflowFromEvent(ctx context.Context, req *protos.RerunWorkflowFromEventRequest) (api.InstanceID, error)
+
 	// AddNewEvent adds a new orchestration event to the specified orchestration instance.
 	AddNewOrchestrationEvent(context.Context, api.InstanceID, *HistoryEvent) error
 

api/protos/orchestrator_service_grpc.pb.go

@@ -28,6 +28,7 @@ type TaskHubClient interface {
 	SuspendOrchestration(ctx context.Context, id api.InstanceID, reason string) error
 	ResumeOrchestration(ctx context.Context, id api.InstanceID, reason string) error
 	PurgeOrchestrationState(ctx context.Context, id api.InstanceID, opts ...api.PurgeOptions) error
+	RerunWorkflowFromEvent(ctx context.Context, source api.InstanceID, eventID uint32, opts ...api.RerunOptions) (api.InstanceID, error)
 }
 
 type backendClient struct {
@@ -257,3 +258,19 @@ func (c *backendClient) PurgeOrchestrationState(ctx context.Context, id api.Inst
 	}
 	return nil
 }
+
+// RerunWorkflowFromEvent reruns a workflow from a specific event ID of some
+// source instance ID. If not given, a random new instance ID will be generated
+// and returned. Can optionally give a new input to the target event ID to
+// rerun from.
+func (c *backendClient) RerunWorkflowFromEvent(ctx context.Context, id api.InstanceID, eventID uint32, opts ...api.RerunOptions) (api.InstanceID, error) {
+	req := &protos.RerunWorkflowFromEventRequest{SourceInstanceID: string(id), EventID: eventID}
+	for _, configure := range opts {
+		if err := configure(req); err != nil {
+			return "", fmt.Errorf("failed to configure rerun request: %w", err)
+		}
+	}
+
+	id, err := c.be.RerunWorkflowFromEvent(ctx, req)
+	return id, err
+}

backend/backend.go

@@ -48,6 +48,7 @@ type Executor interface {
 
 type grpcExecutor struct {
 	protos.UnimplementedTaskHubSidecarServiceServer
+
 	workItemQueue        chan *protos.WorkItem
 	pendingOrchestrators *sync.Map // map[api.InstanceID]*ExecutionResults
 	pendingActivities    *sync.Map // map[string]*activityExecutionResult
@@ -488,6 +489,24 @@ func (g *grpcExecutor) StartInstance(ctx context.Context, req *protos.CreateInst
 	return &protos.CreateInstanceResponse{InstanceId: instanceID}, nil
 }
 
+// RerunWorkflowFromEvent reruns a workflow from a specific event ID of some
+// source instance ID. If not given, a random new instance ID will be
+// generated and returned. Can optionally give a new input to the target
+// event ID to rerun from.
+func (g *grpcExecutor) RerunWorkflowFromEvent(ctx context.Context, req *protos.RerunWorkflowFromEventRequest) (*protos.RerunWorkflowFromEventResponse, error) {
+	newInstanceID, err := g.backend.RerunWorkflowFromEvent(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = g.WaitForInstanceStart(ctx, &protos.GetInstanceRequest{InstanceId: newInstanceID.String()})
+	if err != nil {
+		return nil, err
+	}
+
+	return &protos.RerunWorkflowFromEventResponse{NewInstanceID: newInstanceID.String()}, nil
+}
+
 // TerminateInstance implements protos.TaskHubSidecarServiceServer
 func (g *grpcExecutor) TerminateInstance(ctx context.Context, req *protos.TerminateRequest) (*protos.TerminateResponse, error) {
 	e := &protos.HistoryEvent{
@@ -613,10 +632,6 @@ func (g *grpcExecutor) waitForInstance(ctx context.Context, req *protos.GetInsta
 	return createGetInstanceResponse(req, metadata), nil
 }
 
-// mustEmbedUnimplementedTaskHubSidecarServiceServer implements protos.TaskHubSidecarServiceServer
-func (grpcExecutor) mustEmbedUnimplementedTaskHubSidecarServiceServer() {
-}
-
 func createGetInstanceResponse(req *protos.GetInstanceRequest, metadata *OrchestrationMetadata) *protos.GetInstanceResponse {
 	state := &protos.OrchestrationState{
 		InstanceId:           req.InstanceId,
@@ -635,3 +650,45 @@ func createGetInstanceResponse(req *protos.GetInstanceRequest, metadata *Orchest
 
 	return &protos.GetInstanceResponse{Exists: true, OrchestrationState: state}
 }
+
+func (g *grpcExecutor) AbandonTaskActivityWorkItem(ctx context.Context, req *protos.AbandonActivityTaskRequest) (*protos.AbandonActivityTaskResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+func (g *grpcExecutor) AbandonTaskEntityWorkItem(ctx context.Context, req *protos.AbandonEntityTaskRequest) (*protos.AbandonEntityTaskResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+func (g *grpcExecutor) AbandonTaskOrchestratorWorkItem(ctx context.Context, req *protos.AbandonOrchestrationTaskRequest) (*protos.AbandonOrchestrationTaskResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+func (g *grpcExecutor) CleanEntityStorage(ctx context.Context, req *protos.CleanEntityStorageRequest) (*protos.CleanEntityStorageResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+func (g *grpcExecutor) CompleteEntityTask(ctx context.Context, req *protos.EntityBatchResult) (*protos.CompleteTaskResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+func (g *grpcExecutor) GetEntity(ctx context.Context, req *protos.GetEntityRequest) (*protos.GetEntityResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+func (g *grpcExecutor) QueryEntities(ctx context.Context, req *protos.QueryEntitiesRequest) (*protos.QueryEntitiesResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+func (g *grpcExecutor) RewindInstance(ctx context.Context, req *protos.RewindInstanceRequest) (*protos.RewindInstanceResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+func (g *grpcExecutor) SignalEntity(ctx context.Context, req *protos.SignalEntityRequest) (*protos.SignalEntityResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+func (g *grpcExecutor) StreamInstanceHistory(req *protos.StreamInstanceHistoryRequest, stream protos.TaskHubSidecarService_StreamInstanceHistoryServer) error {
+	return status.Error(codes.Unimplemented, "")
+}
+
+func (grpcExecutor) mustEmbedUnimplementedTaskHubSidecarServiceServer() {}

backend/client.go

@@ -16,6 +16,8 @@ import (
 	"github.com/dapr/durabletask-go/backend"
 	"github.com/dapr/durabletask-go/backend/runtimestate"
 	"github.com/google/uuid"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
 	"google.golang.org/protobuf/types/known/wrapperspb"
@@ -1123,3 +1125,7 @@ func (be *postgresBackend) String() string {
 	connectionURI := fmt.Sprintf("postgresql://%s:%s@%s:%d/%s", be.options.PgOptions.ConnConfig.User, be.options.PgOptions.ConnConfig.Password, be.options.PgOptions.ConnConfig.Host, be.options.PgOptions.ConnConfig.Port, be.options.PgOptions.ConnConfig.Database)
 	return connectionURI
 }
+
+func (be *postgresBackend) RerunWorkflowFromEvent(ctx context.Context, req *backend.RerunWorkflowFromEventRequest) (api.InstanceID, error) {
+	return "", status.Error(codes.Unimplemented, "not implemented")
+}

backend/executor.go

@@ -17,6 +17,8 @@ import (
 	"github.com/dapr/durabletask-go/backend"
 	"github.com/dapr/durabletask-go/backend/runtimestate"
 	"github.com/google/uuid"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
 	"google.golang.org/protobuf/types/known/wrapperspb"
@@ -1108,3 +1110,7 @@ func (be *sqliteBackend) ensureDB() error {
 func (be *sqliteBackend) String() string {
 	return fmt.Sprintf("sqlite::%s", be.options.FilePath)
 }
+
+func (be *sqliteBackend) RerunWorkflowFromEvent(ctx context.Context, req *backend.RerunWorkflowFromEventRequest) (api.InstanceID, error) {
+	return "", status.Error(codes.Unimplemented, "not implemented")
+}

backend/postgres/postgres.go

@@ -210,6 +210,32 @@ func (c *TaskHubGrpcClient) PurgeOrchestrationState(ctx context.Context, id api.
 	return nil
 }
 
+// RerunWorkflowFromEvent reruns a workflow from a specific event ID of some
+// source instance ID. If not given, a random new instance ID will be
+// generated and returned. Can optionally give a new input to the target
+// event ID to rerun from.
+func (c *TaskHubGrpcClient) RerunWorkflowFromEvent(ctx context.Context, id api.InstanceID, eventID uint32, opts ...api.RerunOptions) (api.InstanceID, error) {
+	req := &protos.RerunWorkflowFromEventRequest{
+		SourceInstanceID: string(id),
+		EventID:          eventID,
+	}
+	for _, configure := range opts {
+		if err := configure(req); err != nil {
+			return "", fmt.Errorf("failed to configure rerun request: %w", err)
+		}
+	}
+
+	resp, err := c.client.RerunWorkflowFromEvent(ctx, req)
+	if err != nil {
+		if ctx.Err() != nil {
+			return "", ctx.Err()
+		}
+		return "", err
+	}
+
+	return api.InstanceID(resp.GetNewInstanceID()), nil
+}
+
 func makeGetInstanceRequest(id api.InstanceID, opts []api.FetchOrchestrationMetadataOptions) *protos.GetInstanceRequest {
 	req := &protos.GetInstanceRequest{
 		InstanceId:          string(id),