@effect/cluster: cluster_messages.message_id VARCHAR(255) overflows for DurableQueue deferred envelopes — DurableDeferred.done fails to persist and the workflow never resumes

[IgnisDa]

Summary

SqlMessageStorage creates cluster_messages.message_id as VARCHAR(255), but the dedupe key stored there is Envelope.primaryKeyByAddress — ${entityType}/${entityId}/${tag}/${primaryKey}. For the deferred envelope sent by DurableDeferred.done, that composes to:

Workflow/<workflowName>/<executionId>/deferred/DurableQueue/<queueName>/<sha256-hex(64)>

i.e. ~100+ characters of fixed overhead around the executionId. Once the executionId passes roughly 110–130 characters, every attempt to persist the deferred-done envelope fails with:

PersistenceError: value too long for type character varying(255)

DurableDeferred.done dies after its retries and the awaiting workflow stays Suspended forever.

Long executionIds are easy to hit legitimately: a workflow that runs child workflows must give them deterministic executionIds derived from its own (e.g. `${parentExecutionId}-item-${i}`), otherwise a replay after a durable suspend spawns fresh children on every pass. Two or three levels of that (import → per-item population → per-item queue job) blows past the budget.

Environment

• @effect/cluster 0.59.0
• @effect/workflow 0.18.2
• effect 3.21.4, @effect/sql-pg 0.52.1 (PostgreSQL dialect)
• SingleRunner.layer({ runnerStorage: "sql" })

Root cause

SqlMessageStorage (pg dialect) — dist/esm/SqlMessageStorage.js:

CREATE TABLE IF NOT EXISTS cluster_messages (
  id BIGINT PRIMARY KEY,
  rowid BIGSERIAL,
  message_id VARCHAR(255),
  ...
  UNIQUE (message_id)
)

Envelope.js already anticipates the problem:

export const primaryKeyByAddress = (options) =>
  // hash the entity address to save space?
  `${options.address.entityType}/${options.address.entityId}/${options.tag}/${options.id}`

For a DurableQueue job the deferred's primaryKey (DeferredRpc's primaryKey: ({ name }) => name in ClusterWorkflowEngine.js) is the deferred name built in DurableQueue.process:

const queueName = `DurableQueue/${self.name}`
const id = yield* Activity.idempotencyKey(`${queueName}/${self.idempotencyKey(payload)}`) // sha-256 hex, 64 chars
const deferred = DurableDeferred.make(`${self.deferred.name}/${id}`, ...) // self.deferred.name === `DurableQueue/${self.name}`

So the full message_id is:

Workflow/<workflowName>/<executionId>/deferred/DurableQueue/<queueName>/<64 hex chars>

With e.g. a 20-char workflow name and a 15-char queue name that is ~133 chars before the executionId — an executionId over ~122 chars overflows the column, and the UNIQUE (message_id) insert fails on every retry.

Reproduction sketch

const JobQueue = DurableQueue.make({
  name: "job-queue",
  payload: Schema.Struct({ id: Schema.String }),
  idempotencyKey: ({ id }) => id,
  success: Schema.Void,
})

// slow enough that the workflow durably suspends before the job finishes
const JobWorker = DurableQueue.worker(JobQueue, () => Effect.sleep("1 second"))

const AwaitJob = Workflow.make({
  name: "AwaitJob",
  payload: Schema.Struct({ id: Schema.String }),
  idempotencyKey: ({ id }) => id,
})
const AwaitJobLayer = AwaitJob.toLayer(({ id }) => DurableQueue.process(JobQueue, { id }))

// execute with a long deterministic executionId, as a parent workflow
// deriving child ids would:
const engine = yield* WorkflowEngine
const result = yield* engine.execute(AwaitJob, {
  executionId: `${"parent-execution-id-".repeat(7)}item-0`, // > ~130 chars
  payload: { id: "1" },
})

The workflow suspends, the worker completes, DurableDeferred.done → client(executionId).deferred(...) → the persist of the deferred envelope fails with value too long for type character varying(255), and engine.poll returns Suspended forever.

(Note entity_id is also VARCHAR(255), so executionIds have a hard cap regardless — but message_id overflows far earlier because of the composed prefix/suffix.)

Suggested fix

Either of:

• hash the composed key (as the Envelope.ts comment suggests) — it is only used for dedupe via the UNIQUE constraint, so a digest works; or
• create message_id as TEXT on PostgreSQL.

Workaround we're using

Widening the column right after cluster's migration runs:

const widenClusterMessageIdColumn = Effect.flatMap(
  PgClient.PgClient,
  (sql) => sql`ALTER TABLE cluster_messages ALTER COLUMN message_id TYPE text`,
).pipe(Effect.orDie, Effect.asVoid)

export const WorkflowEngineLive = ClusterWorkflowEngine.layer.pipe(
  Layer.provide(
    SingleRunner.layer({ runnerStorage: "sql" }).pipe(
      Layer.tap(() => widenClusterMessageIdColumn),
    ),
  ),
)

which fully resolves the failures in our e2e suite.