From 46cf346dcae74e0537dde0b48a601170dd246800 Mon Sep 17 00:00:00 2001 From: Javier Aliaga Date: Tue, 14 Jul 2026 11:19:01 +0200 Subject: [PATCH] feat: add enforceUniqueInstanceID option when scheduling workflows Signed-off-by: Javier Aliaga --- api/orchestration.go | 11 ++++++++ backend/backend.go | 6 ++--- backend/client.go | 5 +++- backend/client_router_test.go | 4 +-- backend/executor.go | 8 +++++- backend/postgres/postgres.go | 4 ++- backend/sqlite/sqlite.go | 4 ++- tests/backend_test.go | 6 ++--- tests/grpc/grpc_test.go | 49 +++++++++++++++++++++++++++++++++++ tests/mocks/Backend.go | 12 ++++----- tests/orchestrations_test.go | 49 +++++++++++++++++++++++++++++++++++ 11 files changed, 140 insertions(+), 18 deletions(-) diff --git a/api/orchestration.go b/api/orchestration.go index de701b4..9ca1c70 100644 --- a/api/orchestration.go +++ b/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) { diff --git a/backend/backend.go b/backend/backend.go index c605251..f92a69e 100644 --- a/backend/backend.go +++ b/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 diff --git a/backend/client.go b/backend/client.go index 72bc88f..8c5b489 100644 --- a/backend/client.go +++ b/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) diff --git a/backend/client_router_test.go b/backend/client_router_test.go index f65d4d3..fe8b13b 100644 --- a/backend/client_router_test.go +++ b/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 } diff --git a/backend/executor.go b/backend/executor.go index 9dce9d1..3f3cdb4 100644 --- a/backend/executor.go +++ b/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) } diff --git a/backend/postgres/postgres.go b/backend/postgres/postgres.go index 9cc06b1..2a8ce1f 100644 --- a/backend/postgres/postgres.go +++ b/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) diff --git a/backend/sqlite/sqlite.go b/backend/sqlite/sqlite.go index 606c4c3..d5380c6 100644 --- a/backend/sqlite/sqlite.go +++ b/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) diff --git a/tests/backend_test.go b/tests/backend_test.go index 21e20da..9b23849 100644 --- a/tests/backend_test.go +++ b/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) } diff --git a/tests/grpc/grpc_test.go b/tests/grpc/grpc_test.go index e5df325..5808cd6 100644 --- a/tests/grpc/grpc_test.go +++ b/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) { diff --git a/tests/mocks/Backend.go b/tests/mocks/Backend.go index b2ff0c2..e4adbbe 100644 --- a/tests/mocks/Backend.go +++ b/tests/mocks/Backend.go @@ -452,7 +452,7 @@ func (_c *Backend_CompleteWorkflowTask_Call) RunAndReturn(run func(context.Conte } // CreateWorkflowInstance provides a mock function with given fields: _a0, _a1 -func (_m *Backend) CreateWorkflowInstance(_a0 context.Context, _a1 *protos.HistoryEvent) error { +func (_m *Backend) CreateWorkflowInstance(_a0 context.Context, _a1 *protos.CreateWorkflowInstanceRequest) error { ret := _m.Called(_a0, _a1) if len(ret) == 0 { @@ -460,7 +460,7 @@ func (_m *Backend) CreateWorkflowInstance(_a0 context.Context, _a1 *protos.Histo } var r0 error - if rf, ok := ret.Get(0).(func(context.Context, *protos.HistoryEvent) error); ok { + if rf, ok := ret.Get(0).(func(context.Context, *protos.CreateWorkflowInstanceRequest) error); ok { r0 = rf(_a0, _a1) } else { r0 = ret.Error(0) @@ -476,14 +476,14 @@ type Backend_CreateWorkflowInstance_Call struct { // CreateWorkflowInstance is a helper method to define mock.On call // - _a0 context.Context -// - _a1 *protos.HistoryEvent +// - _a1 *protos.CreateWorkflowInstanceRequest func (_e *Backend_Expecter) CreateWorkflowInstance(_a0 interface{}, _a1 interface{}) *Backend_CreateWorkflowInstance_Call { return &Backend_CreateWorkflowInstance_Call{Call: _e.mock.On("CreateWorkflowInstance", _a0, _a1)} } -func (_c *Backend_CreateWorkflowInstance_Call) Run(run func(_a0 context.Context, _a1 *protos.HistoryEvent)) *Backend_CreateWorkflowInstance_Call { +func (_c *Backend_CreateWorkflowInstance_Call) Run(run func(_a0 context.Context, _a1 *protos.CreateWorkflowInstanceRequest)) *Backend_CreateWorkflowInstance_Call { _c.Call.Run(func(args mock.Arguments) { - run(args[0].(context.Context), args[1].(*protos.HistoryEvent)) + run(args[0].(context.Context), args[1].(*protos.CreateWorkflowInstanceRequest)) }) return _c } @@ -493,7 +493,7 @@ func (_c *Backend_CreateWorkflowInstance_Call) Return(_a0 error) *Backend_Create return _c } -func (_c *Backend_CreateWorkflowInstance_Call) RunAndReturn(run func(context.Context, *protos.HistoryEvent) error) *Backend_CreateWorkflowInstance_Call { +func (_c *Backend_CreateWorkflowInstance_Call) RunAndReturn(run func(context.Context, *protos.CreateWorkflowInstanceRequest) error) *Backend_CreateWorkflowInstance_Call { _c.Call.Return(run) return _c } diff --git a/tests/orchestrations_test.go b/tests/orchestrations_test.go index dda249e..6be2af6 100644 --- a/tests/orchestrations_test.go +++ b/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