feat: add enforceUniqueInstanceID option when scheduling workflows

api/orchestration.go

@@ -102,6 +102,17 @@ func WithStartTime(startTime time.Time) NewWorkflowOptions {
 	}
 }
 
+// WithEnforceUniqueInstanceID configures scheduling to fail with an
+// ALREADY_EXISTS error if an instance with the same ID already exists,
+// whether active or completed. Without it, an existing completed instance
+// is restarted.
+func WithEnforceUniqueInstanceID() NewWorkflowOptions {
+	return func(req *protos.CreateInstanceRequest) error {
+		req.EnforceUniqueInstanceId = true
+		return nil
+	}
+}
+
 // WithFetchPayloads configures whether to load workflow inputs, outputs, and custom status values, which could be large.
 func WithFetchPayloads(fetchPayloads bool) FetchWorkflowMetadataOptions {
 	return func(req *protos.GetInstanceRequest) {

backend/backend.go

@@ -62,9 +62,9 @@ type Backend interface {
 	// Stop stops any background processing done by this backend.
 	Stop(context.Context) error
 
-	// CreateWorkflowInstance creates a new workflow instance with a history event that
-	// wraps a ExecutionStarted event.
-	CreateWorkflowInstance(context.Context, *HistoryEvent) error
+	// CreateWorkflowInstance creates a new workflow instance with a request that
+	// wraps a ExecutionStarted history event.
+	CreateWorkflowInstance(context.Context, *CreateWorkflowInstanceRequest) 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

backend/client.go

@@ -80,7 +80,10 @@ func (c *backendClient) ScheduleNewWorkflow(ctx context.Context, workflow interf
 		},
 		Router: req.GetRouter(),
 	}
-	if err := c.be.CreateWorkflowInstance(ctx, e); err != nil {
+	if err := c.be.CreateWorkflowInstance(ctx, &CreateWorkflowInstanceRequest{
+		StartEvent:              e,
+		EnforceUniqueInstanceId: req.EnforceUniqueInstanceId,
+	}); err != nil {
 		span.RecordError(err)
 		span.SetStatus(codes.Error, err.Error())
 		return api.EmptyInstanceID, fmt.Errorf("failed to start workflow: %w", err)

backend/client_router_test.go

@@ -36,8 +36,8 @@ type fakeRouterBackend struct {
 	watchRouter  *protos.TaskRouter
 }
 
-func (f *fakeRouterBackend) CreateWorkflowInstance(_ context.Context, e *HistoryEvent) error {
-	f.createdEvent = e
+func (f *fakeRouterBackend) CreateWorkflowInstance(_ context.Context, req *CreateWorkflowInstanceRequest) error {
+	f.createdEvent = req.GetStartEvent()
 	return nil
 }
 

backend/executor.go

@@ -567,7 +567,13 @@ func (g *grpcExecutor) StartInstance(ctx context.Context, req *protos.CreateInst
 		},
 		Router: req.GetRouter(),
 	}
-	if err := g.backend.CreateWorkflowInstance(ctx, e); err != nil {
+	if err := g.backend.CreateWorkflowInstance(ctx, &CreateWorkflowInstanceRequest{
+		StartEvent:              e,
+		EnforceUniqueInstanceId: req.EnforceUniqueInstanceId,
+	}); err != nil {
+		if errors.Is(err, api.ErrDuplicateInstance) {
+			return nil, status.Error(codes.AlreadyExists, err.Error())
+		}
 		return nil, fmt.Errorf("failed to create workflow instance: %w", err)
 	}
 

backend/postgres/postgres.go

@@ -540,11 +540,13 @@ func (be *postgresBackend) CompleteWorkflowWorkItem(ctx context.Context, wi *bac
 }
 
 // CreateWorkflowInstance implements backend.Backend
-func (be *postgresBackend) CreateWorkflowInstance(ctx context.Context, e *backend.HistoryEvent) error {
+func (be *postgresBackend) CreateWorkflowInstance(ctx context.Context, req *backend.CreateWorkflowInstanceRequest) error {
 	if err := be.ensureDB(); err != nil {
 		return err
 	}
 
+	e := req.GetStartEvent()
+
 	tx, err := be.db.BeginTx(ctx, pgx.TxOptions{})
 	if err != nil {
 		return fmt.Errorf("failed to start transaction: %w", err)

backend/sqlite/sqlite.go

@@ -411,11 +411,13 @@ func (be *sqliteBackend) CompleteWorkflowWorkItem(ctx context.Context, wi *backe
 }
 
 // CreateWorkflowInstance implements backend.Backend
-func (be *sqliteBackend) CreateWorkflowInstance(ctx context.Context, e *backend.HistoryEvent) error {
+func (be *sqliteBackend) CreateWorkflowInstance(ctx context.Context, req *backend.CreateWorkflowInstanceRequest) error {
 	if err := be.ensureDB(); err != nil {
 		return err
 	}
 
+	e := req.GetStartEvent()
+
 	tx, err := be.db.BeginTx(ctx, nil)
 	if err != nil {
 		return fmt.Errorf("failed to start transaction: %w", err)

tests/backend_test.go

@@ -435,7 +435,7 @@ func Test_GetWorkflowMetadata_StartedAt(t *testing.T) {
 				},
 			},
 		}
-		if !assert.NoError(t, be.CreateWorkflowInstance(ctx, e)) {
+		if !assert.NoError(t, be.CreateWorkflowInstance(ctx, &backend.CreateWorkflowInstanceRequest{StartEvent: e})) {
 			continue
 		}
 
@@ -603,7 +603,7 @@ func createWorkflowInstance(t assert.TestingT, be backend.Backend, instanceID st
 			},
 		},
 	}
-	err := be.CreateWorkflowInstance(ctx, e)
+	err := be.CreateWorkflowInstance(ctx, &backend.CreateWorkflowInstanceRequest{StartEvent: e})
 	return assert.NoError(t, err)
 }
 
@@ -622,7 +622,7 @@ func createChildWorkflowInstance(t assert.TestingT, be backend.Backend, instance
 			},
 		},
 	}
-	err := be.CreateWorkflowInstance(ctx, e)
+	err := be.CreateWorkflowInstance(ctx, &backend.CreateWorkflowInstanceRequest{StartEvent: e})
 	return assert.NoError(t, err)
 }
 

tests/grpc/grpc_test.go

@@ -15,7 +15,9 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
 
 	"github.com/dapr/durabletask-go/api"
 	"github.com/dapr/durabletask-go/api/protos"
@@ -341,6 +343,53 @@ func Test_Grpc_ReuseInstanceIDError(t *testing.T) {
 	}
 }
 
+func Test_Grpc_EnforceUniqueInstanceID(t *testing.T) {
+	r := task.NewTaskRegistry()
+	r.AddWorkflowN("SingleActivity", func(ctx *task.WorkflowContext) (any, error) {
+		var input string
+		if err := ctx.GetInput(&input); err != nil {
+			return nil, err
+		}
+		var output string
+		err := ctx.CallActivity("SayHello", task.WithActivityInput(input)).Await(&output)
+		return output, err
+	})
+	r.AddActivityN("SayHello", func(ctx task.ActivityContext) (any, error) {
+		var name string
+		if err := ctx.GetInput(&name); err != nil {
+			return nil, err
+		}
+		return fmt.Sprintf("Hello, %s!", name), nil
+	})
+
+	cancelListener := startGrpcListener(t, r)
+	defer cancelListener()
+	instanceID := api.InstanceID("GRPC_ENFORCE_UNIQUE_INSTANCE_ID")
+
+	id, err := grpcClient.ScheduleNewWorkflow(ctx, "SingleActivity", api.WithInput("世界"), api.WithInstanceID(instanceID))
+	require.NoError(t, err)
+
+	// Scheduling again with the flag while the instance is active fails
+	// with an ALREADY_EXISTS error
+	_, err = grpcClient.ScheduleNewWorkflow(ctx, "SingleActivity", api.WithInput("World"), api.WithInstanceID(id), api.WithEnforceUniqueInstanceID())
+	require.Error(t, err)
+	assert.Equal(t, codes.AlreadyExists, status.Code(err), err)
+
+	metadata, err := grpcClient.WaitForWorkflowCompletion(ctx, id)
+	require.NoError(t, err)
+	assert.Equal(t, `"Hello, 世界!"`, metadata.Output.GetValue())
+
+	// Scheduling again with the flag after completion also fails and does
+	// not restart the completed instance
+	_, err = grpcClient.ScheduleNewWorkflow(ctx, "SingleActivity", api.WithInput("World"), api.WithInstanceID(id), api.WithEnforceUniqueInstanceID())
+	require.Error(t, err)
+	assert.Equal(t, codes.AlreadyExists, status.Code(err), err)
+
+	metadata, err = grpcClient.WaitForWorkflowCompletion(ctx, id)
+	require.NoError(t, err)
+	assert.Equal(t, `"Hello, 世界!"`, metadata.Output.GetValue())
+}
+
 func Test_Grpc_ActivityRetries(t *testing.T) {
 	r := task.NewTaskRegistry()
 	r.AddWorkflowN("ActivityRetries", func(ctx *task.WorkflowContext) (any, error) {

tests/orchestrations_test.go

@@ -1511,6 +1511,55 @@ func Test_SingleActivity_ReuseInstanceIDError(t *testing.T) {
 	}
 }
 
+func Test_SingleActivity_EnforceUniqueInstanceID(t *testing.T) {
+	// Registration
+	r := task.NewTaskRegistry()
+	r.AddWorkflowN("SingleActivity", func(ctx *task.WorkflowContext) (any, error) {
+		var input string
+		if err := ctx.GetInput(&input); err != nil {
+			return nil, err
+		}
+		var output string
+		err := ctx.CallActivity("SayHello", task.WithActivityInput(input)).Await(&output)
+		return output, err
+	})
+	r.AddActivityN("SayHello", func(ctx task.ActivityContext) (any, error) {
+		var name string
+		if err := ctx.GetInput(&name); err != nil {
+			return nil, err
+		}
+		return fmt.Sprintf("Hello, %s!", name), nil
+	})
+
+	// Initialization
+	ctx := context.Background()
+	client, worker := initTaskHubWorker(ctx, r)
+	defer worker.Shutdown(ctx)
+
+	instanceID := api.InstanceID("ENFORCE_UNIQUE_INSTANCE_ID")
+
+	// Run the workflow
+	id, err := client.ScheduleNewWorkflow(ctx, "SingleActivity", api.WithInput("世界"), api.WithInstanceID(instanceID))
+	require.NoError(t, err)
+
+	// Scheduling again with the flag while the instance is active fails
+	_, err = client.ScheduleNewWorkflow(ctx, "SingleActivity", api.WithInput("World"), api.WithInstanceID(id), api.WithEnforceUniqueInstanceID())
+	require.ErrorIs(t, err, api.ErrDuplicateInstance)
+
+	metadata, err := client.WaitForWorkflowCompletion(ctx, id)
+	require.NoError(t, err)
+	assert.Equal(t, `"Hello, 世界!"`, metadata.Output.GetValue())
+
+	// Scheduling again with the flag after completion also fails and does
+	// not restart the completed instance
+	_, err = client.ScheduleNewWorkflow(ctx, "SingleActivity", api.WithInput("World"), api.WithInstanceID(id), api.WithEnforceUniqueInstanceID())
+	require.ErrorIs(t, err, api.ErrDuplicateInstance)
+
+	metadata, err = client.WaitForWorkflowCompletion(ctx, id)
+	require.NoError(t, err)
+	assert.Equal(t, `"Hello, 世界!"`, metadata.Output.GetValue())
+}
+
 func Test_TaskExecutionId(t *testing.T) {
 	t.Run("SingleActivityWithRetry", func(t *testing.T) {
 		// Registration