Hash SQL message deduplication keys to prevent message_id overflow

.changeset/hash-sql-message-dedupe-keys.md

@@ -0,0 +1,11 @@
+---
+"effect": patch
+"@effect/platform-node": patch
+"@effect/platform-bun": patch
+---
+
+unstable/cluster: hash over-length SQL message deduplication keys to prevent `message_id` overflow, closes #6317.
+
+The composed request deduplication key (`entityType/entityId/tag/primaryKey`) can legally exceed the 255-character `message_id` column — the address columns alone allow 458 characters before the RPC primary key is appended. `SqlMessageStorage` now stores a SHA-256 digest (64 hex characters) of the composed key in the unique `message_id` column when the key exceeds 255 characters, so keys of any length work on PostgreSQL, MySQL, MSSQL, and SQLite. Keys that fit are stored as plaintext, byte-compatible with rows written by previous versions, so existing deployments keep deduplicating with no migration or schema change.
+
+`SqlMessageStorage.layer`/`layerWith` (and consequently `SingleRunner.layer`) now require `Crypto.Crypto`. The Node and Bun cluster convenience layers provide the platform Crypto implementation internally, so their requirements are unchanged.

packages/effect/src/unstable/cluster/Envelope.ts

@@ -429,5 +429,6 @@ export const primaryKeyByAddress = (options: {
   readonly tag: string
   readonly id: string
 }): string =>
-  // hash the entity address to save space?
+  // storage drivers with fixed-width key columns (e.g. SqlMessageStorage)
+  // hash this composed key at their own boundary
   `${options.address.entityType}/${options.address.entityId}/${options.tag}/${options.id}`

packages/effect/src/unstable/cluster/SingleRunner.ts

@@ -12,6 +12,7 @@
  */
 import * as Layer from "effect/Layer"
 import type { ConfigError } from "../../Config.ts"
+import type * as Crypto from "../../Crypto.ts"
 import type * as SqlClient from "../sql/SqlClient.ts"
 import type * as MessageStorage from "./MessageStorage.ts"
 import * as RunnerHealth from "./RunnerHealth.ts"
@@ -42,7 +43,8 @@ import * as SqlRunnerStorage from "./SqlRunnerStorage.ts"
  * **Gotchas**
  *
  * - Even when `runnerStorage` is `"memory"`, message storage remains
- *   SQL-backed, so callers must still provide `SqlClient`.
+ *   SQL-backed, so callers must still provide `SqlClient` and `Crypto.Crypto`
+ *   (used to hash over-length message deduplication keys).
  * - Runner communication and runner health are no-op services, so this layer is
  *   for single-process use rather than multi-runner coordination.
  *
@@ -62,7 +64,7 @@ export const layer = (options?: {
   | Runners.Runners
   | MessageStorage.MessageStorage,
   ConfigError,
-  SqlClient.SqlClient
+  SqlClient.SqlClient | Crypto.Crypto
 > =>
   Sharding.layer.pipe(
     Layer.provideMerge(Runners.layerNoop),

packages/effect/src/unstable/cluster/SqlMessageStorage.ts

@@ -8,13 +8,21 @@
  * storage constructor, layers, migrations, optional table prefixes, and the row
  * mapping needed by encoded message storage.
  *
+ * Request deduplication keys that exceed the 255-character `message_id`
+ * column are hashed with the `Crypto` service before they are written, so
+ * composed keys of any length are supported; shorter keys are stored as
+ * plaintext, byte-compatible with rows written by previous versions.
+ *
  * @since 4.0.0
  */
 // eslint-disable effect/no-bigint-literals
 import * as Arr from "../../Array.ts"
+import * as Crypto from "../../Crypto.ts"
 import * as Effect from "../../Effect.ts"
+import * as Encoding from "../../Encoding.ts"
 import * as Layer from "../../Layer.ts"
 import * as Option from "../../Option.ts"
+import type * as PlatformError from "../../PlatformError.ts"
 import * as Schedule from "../../Schedule.ts"
 import * as Migrator from "../sql/Migrator.ts"
 import * as SqlClient from "../sql/SqlClient.ts"
@@ -61,9 +69,10 @@ export const make: (options?: {
 }) => Effect.Effect<
   MessageStorage.MessageStorage["Service"],
   never,
-  SqlClient.SqlClient | Snowflake.Generator
+  SqlClient.SqlClient | Snowflake.Generator | Crypto.Crypto
 > = Effect.fnUntraced(function*(options) {
   const sql = (yield* SqlClient.SqlClient).withoutTransforms()
+  const crypto = yield* Crypto.Crypto
   const prefix = options?.prefix ?? "cluster"
   const table = (name: string) => `${prefix}_${name}`
 
@@ -84,6 +93,30 @@ export const make: (options?: {
   const repliesTable = table("replies")
   const repliesTableSql = sql(repliesTable)
 
+  // The composed primary key (`entityType/entityId/tag/id`) can legally exceed
+  // the 255-character `message_id` column: entity_type(150) + entity_id(255) +
+  // tag(50) alone total 458 characters before the RPC primary key is appended.
+  // Keys that fit are stored as-is, keeping `message_id` byte-compatible with
+  // rows written by previous versions; longer keys are stored as a SHA-256
+  // digest (64 hex characters, collision probability negligible at a 2^128
+  // birthday bound). Digests never contain "/" while composed keys always do,
+  // so the two encodings cannot collide.
+  const encoder = new TextEncoder()
+  const messageIdForPrimaryKey = (primaryKey: string): Effect.Effect<string, PlatformError.PlatformError> =>
+    primaryKey.length <= 255
+      ? Effect.succeed(primaryKey)
+      : Effect.map(crypto.digest("SHA-256", encoder.encode(primaryKey)), Encoding.encodeHex)
+
+  const messageIdEnforcesWidth = sql.onDialectOrElse({
+    mssql: () => true,
+    mysql: () => true,
+    pg: () => true,
+    orElse: () => false
+  })
+  // sqlite's TEXT message_id column stored over-length plaintext keys before
+  // digests were introduced; those legacy rows need a plaintext fallback read
+  const mayHaveLegacyRow = (primaryKey: string): boolean => !messageIdEnforcesWidth && primaryKey.length > 255
+
   const envelopeToRow = (
     envelope: Envelope.Encoded,
     message_id: string | null,
@@ -238,6 +271,14 @@ export const make: (options?: {
   const sqlFalse = sql.literal(supportsBooleans ? "FALSE" : "0")
   const sqlTrue = sql.literal(supportsBooleans ? "TRUE" : "1")
 
+  const selectByMessageId = (message_id: string): Effect.Effect<ReadonlyArray<Row>, SqlError> =>
+    sql`
+      SELECT m.id, r.id as reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence
+      FROM ${messagesTableSql} m
+      LEFT JOIN ${repliesTableSql} r ON r.id = m.last_reply_id
+      WHERE m.message_id = ${message_id}
+    `
+
   const insertEnvelope: (
     row: MessageRow,
     message_id: string
@@ -250,12 +291,7 @@ export const make: (options?: {
       `.pipe(Effect.flatMap((rows) => {
         // inserted a new row
         if (rows.length > 0) return Effect.succeed([])
-        return sql`
-          SELECT m.id, r.id as reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence
-          FROM ${messagesTableSql} m
-          LEFT JOIN ${repliesTableSql} r ON r.id = m.last_reply_id
-          WHERE m.message_id = ${message_id}
-        `
+        return selectByMessageId(message_id)
       })),
     mysql: () => (row, message_id) =>
       Effect.flatMap(
@@ -264,12 +300,7 @@ export const make: (options?: {
           if (row.affectedRows > 0) {
             return Effect.succeed([])
           }
-          return sql`
-            SELECT m.id, r.id as reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence
-            FROM ${messagesTableSql} m
-            LEFT JOIN ${repliesTableSql} r ON r.id = m.last_reply_id
-            WHERE m.message_id = ${message_id}
-          `
+          return selectByMessageId(message_id)
         }
       ),
     mssql: () => (row, message_id) =>
@@ -311,12 +342,7 @@ export const make: (options?: {
           END as reply_sequence;
       `,
     orElse: () => (row, message_id) =>
-      sql`
-        SELECT m.id, r.id as reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence
-        FROM ${messagesTableSql} m
-        LEFT JOIN ${repliesTableSql} r ON r.id = m.last_reply_id
-        WHERE m.message_id = ${message_id}
-      `.pipe(
+      selectByMessageId(message_id).pipe(
         Effect.tap(sql`INSERT OR IGNORE INTO ${messagesTableSql} ${sql.insert(row)}`),
         sql.withTransaction,
         Effect.retry({ times: 3 })
@@ -406,18 +432,30 @@ export const make: (options?: {
   return yield* MessageStorage.makeEncoded({
     saveEnvelope: ({ deliverAt, envelope, primaryKey }) =>
       Effect.suspend(() => {
-        const row = envelopeToRow(envelope, primaryKey, deliverAt)
-        let insert = primaryKey
-          ? insertEnvelope(row, primaryKey)
-          : Effect.as(sql`INSERT INTO ${messagesTableSql} ${sql.insert(row)}`.unprepared, [])
-        if (envelope._tag === "AckChunk") {
-          insert = sql`UPDATE ${repliesTableSql} SET acked = ${sqlTrue} WHERE id = ${envelope.replyId}`.pipe(
-            Effect.andThen(
-              sql`UPDATE ${messagesTableSql} SET processed = ${sqlTrue} WHERE processed = ${sqlFalse} AND request_id = ${envelope.requestId} AND kind = ${messageKindAckChunk}`
-            ),
-            Effect.andThen(insert),
-            sql.withTransaction
-          )
+        let insert: Effect.Effect<ReadonlyArray<Row>, SqlError | PlatformError.PlatformError>
+        if (primaryKey !== null) {
+          insert = Effect.flatMap(messageIdForPrimaryKey(primaryKey), (messageId) => {
+            const row = envelopeToRow(envelope, messageId, deliverAt)
+            if (!mayHaveLegacyRow(primaryKey)) {
+              return insertEnvelope(row, messageId)
+            }
+            return Effect.flatMap(
+              selectByMessageId(primaryKey),
+              (rows) => rows.length > 0 ? Effect.succeed(rows) : insertEnvelope(row, messageId)
+            )
+          })
+        } else {
+          const row = envelopeToRow(envelope, null, deliverAt)
+          insert = Effect.as(sql`INSERT INTO ${messagesTableSql} ${sql.insert(row)}`.unprepared, [])
+          if (envelope._tag === "AckChunk") {
+            insert = sql`UPDATE ${repliesTableSql} SET acked = ${sqlTrue} WHERE id = ${envelope.replyId}`.pipe(
+              Effect.andThen(
+                sql`UPDATE ${messagesTableSql} SET processed = ${sqlTrue} WHERE processed = ${sqlFalse} AND request_id = ${envelope.requestId} AND kind = ${messageKindAckChunk}`
+              ),
+              Effect.andThen(insert),
+              sql.withTransaction
+            )
+          }
         }
         return insert.pipe(
           Effect.map((rows) => {
@@ -488,7 +526,15 @@ export const make: (options?: {
     ),
 
     requestIdForPrimaryKey: (primaryKey) =>
-      sql<{ id: string | bigint }>`SELECT id FROM ${messagesTableSql} WHERE message_id = ${primaryKey}`.pipe(
+      messageIdForPrimaryKey(primaryKey).pipe(
+        Effect.flatMap((messageId) =>
+          sql<{ id: string | bigint }>`SELECT id FROM ${messagesTableSql} WHERE message_id = ${messageId}`
+        ),
+        Effect.flatMap((rows) =>
+          rows.length === 0 && mayHaveLegacyRow(primaryKey)
+            ? sql<{ id: string | bigint }>`SELECT id FROM ${messagesTableSql} WHERE message_id = ${primaryKey}`
+            : Effect.succeed(rows)
+        ),
         Effect.map((rows) => Option.map(Option.fromNullishOr(rows[0]?.id), Snowflake.Snowflake)),
         Effect.provideService(SqlClient.SafeIntegers, true),
         PersistenceError.refail,
@@ -645,7 +691,9 @@ export const make: (options?: {
  *
  * The layer runs the SQL migrations through `make`, provides `MessageStorage`,
  * and supplies `Snowflake.layerGenerator` internally. Callers still provide
- * `SqlClient` and `ShardingConfig`.
+ * `SqlClient`, `ShardingConfig`, and `Crypto.Crypto`, which is used to hash
+ * message deduplication keys that would overflow the fixed-width
+ * `message_id` column.
  *
  * **Gotchas**
  *
@@ -662,7 +710,7 @@ export const make: (options?: {
 export const layer: Layer.Layer<
   MessageStorage.MessageStorage,
   never,
-  SqlClient.SqlClient | ShardingConfig
+  SqlClient.SqlClient | ShardingConfig | Crypto.Crypto
 > = Layer.effect(MessageStorage.MessageStorage, make()).pipe(
   Layer.provide(Snowflake.layerGenerator)
 )
@@ -675,7 +723,7 @@ export const layer: Layer.Layer<
  */
 export const layerWith = (options: {
   readonly prefix?: string | undefined
-}): Layer.Layer<MessageStorage.MessageStorage, never, SqlClient.SqlClient | ShardingConfig> =>
+}): Layer.Layer<MessageStorage.MessageStorage, never, SqlClient.SqlClient | ShardingConfig | Crypto.Crypto> =>
   Layer.effect(MessageStorage.MessageStorage, make(options)).pipe(
     Layer.provide(Snowflake.layerGenerator)
   )

packages/platform-bun/src/BunClusterHttp.ts

@@ -29,6 +29,7 @@ import type { ServeError } from "effect/unstable/http/HttpServerError"
 import * as RpcSerialization from "effect/unstable/rpc/RpcSerialization"
 import type { SqlClient } from "effect/unstable/sql/SqlClient"
 import { layerK8sHttpClient } from "./BunClusterSocket.ts"
+import * as BunCrypto from "./BunCrypto.ts"
 import * as BunFileSystem from "./BunFileSystem.ts"
 import * as BunHttpServer from "./BunHttpServer.ts"
 import type { BunServices } from "./BunServices.ts"
@@ -157,7 +158,7 @@ export const layer = <
         ? MessageStorage.layerNoop
         : options?.storage === "byo"
         ? Layer.empty
-        : Layer.orDie(SqlMessageStorage.layer)
+        : Layer.orDie(SqlMessageStorage.layer).pipe(Layer.provide(BunCrypto.layer))
     ),
     Layer.provide(
       options?.storage === "local"

packages/platform-bun/src/BunClusterSocket.ts

@@ -28,6 +28,7 @@ import * as FetchHttpClient from "effect/unstable/http/FetchHttpClient"
 import * as RpcSerialization from "effect/unstable/rpc/RpcSerialization"
 import type * as SocketServer from "effect/unstable/socket/SocketServer"
 import type { SqlClient } from "effect/unstable/sql/SqlClient"
+import * as BunCrypto from "./BunCrypto.ts"
 import * as BunFileSystem from "./BunFileSystem.ts"
 
 export {
@@ -109,7 +110,7 @@ export const layer = <
         ? MessageStorage.layerNoop
         : options?.storage === "byo"
         ? Layer.empty
-        : Layer.orDie(SqlMessageStorage.layer)
+        : Layer.orDie(SqlMessageStorage.layer).pipe(Layer.provide(BunCrypto.layer))
     ),
     Layer.provide(
       options?.storage === "local"

packages/platform-node/src/NodeClusterHttp.ts

@@ -31,6 +31,7 @@ import * as RpcSerialization from "effect/unstable/rpc/RpcSerialization"
 import type { SqlClient } from "effect/unstable/sql/SqlClient"
 import { createServer } from "node:http"
 import { layerK8sHttpClient } from "./NodeClusterSocket.ts"
+import * as NodeCrypto from "./NodeCrypto.ts"
 import * as NodeHttpClient from "./NodeHttpClient.ts"
 import * as NodeHttpServer from "./NodeHttpServer.ts"
 import type { NodeServices } from "./NodeServices.ts"
@@ -115,7 +116,7 @@ export const layer = <
         ? MessageStorage.layerNoop
         : options?.storage === "byo"
         ? Layer.empty
-        : Layer.orDie(SqlMessageStorage.layer)
+        : Layer.orDie(SqlMessageStorage.layer).pipe(Layer.provide(NodeCrypto.layer))
     ),
     Layer.provide(
       options?.storage === "local"

packages/platform-node/src/NodeClusterSocket.ts

@@ -28,6 +28,7 @@ import * as SqlRunnerStorage from "effect/unstable/cluster/SqlRunnerStorage"
 import * as RpcSerialization from "effect/unstable/rpc/RpcSerialization"
 import type * as SocketServer from "effect/unstable/socket/SocketServer"
 import type { SqlClient } from "effect/unstable/sql/SqlClient"
+import * as NodeCrypto from "./NodeCrypto.ts"
 import * as NodeFileSystem from "./NodeFileSystem.ts"
 import * as NodeHttpClient from "./NodeHttpClient.ts"
 import * as Undici from "./Undici.ts"
@@ -113,7 +114,7 @@ export const layer = <
         ? MessageStorage.layerNoop
         : options?.storage === "byo"
         ? Layer.empty
-        : Layer.orDie(SqlMessageStorage.layer)
+        : Layer.orDie(SqlMessageStorage.layer).pipe(Layer.provide(NodeCrypto.layer))
     ),
     Layer.provide(
       options?.storage === "local"

packages/platform-node/test/cluster/MessageStorageTest.ts

@@ -136,6 +136,13 @@ export class PrimaryKeyTest extends Rpc.make("PrimaryKeyTest", {
   primaryKey: (value) => value.id.toString()
 }) {}
 
+export class LongKeyRpc extends Rpc.make("LongKeyRpc", {
+  payload: {
+    id: Schema.String
+  },
+  primaryKey: (value) => value.id
+}) {}
+
 export class StreamRpc extends Rpc.make("StreamTest", {
   success: RpcSchema.Stream(Schema.Void, Schema.Never),
   payload: {