Postgres backend can create duplicate active instances under concurrent CreateWorkflowInstance calls

[meoyawn]

Postgres backend can create duplicate active instances under concurrent CreateWorkflowInstance calls

Summary

The Postgres backend currently prevents duplicate active workflow instances with the same instance_id using a read-before-insert check inside a READ COMMITTED transaction. This catches sequential duplicate starts, but it is not a hard guarantee under concurrent starts because the schema does not enforce uniqueness for active instance_ids.

This appears related to:

• Prevent creating multiple workflow instances with the same instance id #260
• Prevent multiple active instances with the same id #288

PR #288 establishes the intended invariant: only one active workflow instance per instance ID. However, on the current Postgres backend, that invariant can still be violated by two concurrent CreateWorkflowInstance calls with the same InstanceID and different execution IDs.

I checked current master:

8ef4f86f45f44c8a22b1ce2138d329ddef3aaf1c

Current behavior

CreateWorkflowInstance starts a transaction at sql.LevelReadCommitted:

go-workflows/backend/postgres/postgres.go

Lines 179 to 184 in 8ef4f86

	func (pb *postgresBackend) CreateWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error {
	tx, err := pb.db.BeginTx(ctx, &sql.TxOptions{
	Isolation: sql.LevelReadCommitted,
	})
	if err != nil {
	return fmt.Errorf("starting transaction: %w", err)

createInstance then checks for an existing active instance:

go-workflows/backend/postgres/postgres.go

Lines 447 to 459 in 8ef4f86

	func createInstance(ctx context.Context, tx *sql.Tx, queue workflow.Queue, wfi *workflow.Instance, metadata *workflow.Metadata) error {
	// Check for existing instance
	err := tx.QueryRowContext(
	ctx,
	"SELECT 1 FROM instances WHERE instance_id = $1 AND state = $2 LIMIT 1",
	wfi.InstanceID,
	core.WorkflowInstanceStateActive).
	Scan(new(int))
	if err == nil {
	return backend.ErrInstanceAlreadyExists
	}
	if !errors.Is(err, sql.ErrNoRows) {
	return err

and inserts the new active instance:

go-workflows/backend/postgres/postgres.go

Lines 475 to 488 in 8ef4f86

	_, err = tx.ExecContext(
	ctx,
	"INSERT INTO instances (queue, instance_id, execution_id, parent_instance_id, parent_execution_id, parent_schedule_event_id, metadata, state) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
	string(queue),
	wfi.InstanceID,
	wfi.ExecutionID,
	parentInstanceID,
	parentExecutionID,
	parentEventID,
	metadataJson,
	core.WorkflowInstanceStateActive,
	)
	if err != nil {
	return fmt.Errorf("inserting workflow instance: %w", err)

The migration only enforces uniqueness on (instance_id, execution_id):

go-workflows/backend/postgres/db/migrations/000001_initial.up.sql

Line 20 in 8ef4f86

CREATE UNIQUE INDEX idx_instances_instance_id_execution_id on instances (instance_id, execution_id);

That index does not prevent multiple active rows for the same instance_id when they have different execution_ids.

Race

With two callers starting the same deterministic instance ID at the same time:

T1: BEGIN READ COMMITTED
T2: BEGIN READ COMMITTED

T1: SELECT 1 FROM instances WHERE instance_id = 'rss-import-abc' AND state = active LIMIT 1
    -> no committed row

T2: SELECT 1 FROM instances WHERE instance_id = 'rss-import-abc' AND state = active LIMIT 1
    -> no committed row

T1: INSERT INTO instances (instance_id='rss-import-abc', execution_id='exec-a', state=active)
T2: INSERT INTO instances (instance_id='rss-import-abc', execution_id='exec-b', state=active)

T1: COMMIT
T2: COMMIT

Both inserts satisfy the current (instance_id, execution_id) unique index. The result is two active instances with the same instance_id.

Why the existing test does not catch this

The existing backend test covers sequential duplicate creation:

go-workflows/backend/test/backendtest.go

Lines 42 to 65 in 8ef4f86

	name: "CreateWorkflowInstance_SameInstanceIDErrors",
	f: func(t *testing.T, ctx context.Context, b backend.Backend) {
	instanceID := uuid.NewString()
	executionID1 := uuid.NewString()
	executionID2 := uuid.NewString()
	err := b.CreateWorkflowInstance(ctx,
	core.NewWorkflowInstance(instanceID, executionID1),
	history.NewHistoryEvent(1, time.Now(), history.EventType_WorkflowExecutionStarted, &history.ExecutionStartedAttributes{
	Queue: workflow.QueueDefault,
	}),
	)
	require.NoError(t, err)
	err = b.CreateWorkflowInstance(
	ctx,
	core.NewWorkflowInstance(instanceID, executionID2),
	history.NewHistoryEvent(1, time.Now(), history.EventType_WorkflowExecutionStarted, &history.ExecutionStartedAttributes{
	Queue: workflow.QueueDefault,
	}),
	)
	require.Error(t, err)
	require.ErrorIs(t, err, backend.ErrInstanceAlreadyExists)
	},

That test is valid, but it does not exercise two concurrent transactions that both observe no active row before either commits.

Why this matters

WorkflowInstanceOptions.InstanceID is useful as an application-level idempotency key. For example, an API may derive a deterministic instance ID from a business key such as (teamID, canonicalURL) and rely on ErrInstanceAlreadyExists to ensure that two identical submissions attach to the same workflow instead of starting duplicate workflows.

With the current Postgres backend, this works for sequential calls but not as a hard concurrency boundary.

It also affects signaling semantics: SignalWorkflow resolves an active execution by instance_id:

go-workflows/backend/postgres/postgres.go

Line 505 in 8ef4f86

res := tx.QueryRowContext(ctx, "SELECT execution_id FROM instances WHERE instance_id = $1 AND state = $2 LIMIT 1", instanceID, core.WorkflowInstanceStateActive)

If two active executions exist for the same instance_id, a signal can target an arbitrary one.

Possible fixes

Option A: partial unique index

Add a partial unique index for active instances:

CREATE UNIQUE INDEX idx_instances_one_active_per_instance_id
ON instances (instance_id)
WHERE state = 0;

0 is currently core.WorkflowInstanceStateActive.

This provides the strongest database guarantee. The insert path would also need to translate the unique violation into backend.ErrInstanceAlreadyExists.

The main thing to verify is ContinueAsNew ordering. If a continued-as-new execution temporarily requires the old and new execution for the same instance_id to both be active in one transaction, this index would conflict. If the old execution is marked ContinuedAsNew before inserting the next active execution, the partial index should work.

Option B: transaction-scoped advisory lock

Serialize creation by instance_id before the existing check:

SELECT pg_advisory_xact_lock(hashtextextended($1, 0));

where $1 is instance_id.

Then run the existing active-instance check and insert. This avoids requiring a pre-existing row to lock and avoids needing to change the uniqueness model around completed/continued executions.

This is less declarative than a unique index, but it may be the least invasive fix for the Postgres backend.

Option C: stronger isolation

Using SERIALIZABLE might also surface a serialization failure for this pattern, but that would require callers/backend code to handle retry behavior. A direct uniqueness constraint or per-instance advisory lock seems clearer.

Expected behavior

For concurrent CreateWorkflowInstance calls with the same instance_id and different execution_ids:

• exactly one call should create an active instance
• all others should return backend.ErrInstanceAlreadyExists
• the database should not be able to contain two active rows with the same instance_id

Suggested regression test

Add a Postgres-specific concurrency test that starts many goroutines calling CreateWorkflowInstance with the same instance_id and unique execution IDs, then asserts:

• one call succeeds
• all other calls return backend.ErrInstanceAlreadyExists
• querying instances shows exactly one active row for that instance_id

The current shared backend test should remain, but the Postgres backend needs a concurrent version because this is a database isolation/schema issue.