From 4d0f35e06cc635c5b68c0010d533c14087eed007 Mon Sep 17 00:00:00 2001
From: Tim Smart <hello@timsmart.co>
Date: Sat, 25 Jul 2026 22:42:06 +1200
Subject: [PATCH 1/5] Hash SQL message deduplication keys to prevent message_id
 overflow

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .changeset/hash-sql-message-dedupe-keys.md    |  13 ++
 .../effect/src/unstable/cluster/Envelope.ts   |   3 +-
 .../src/unstable/cluster/SingleRunner.ts      |   6 +-
 .../src/unstable/cluster/SqlMessageStorage.ts | 131 +++++++++++++++---
 packages/platform-bun/src/BunClusterHttp.ts   |   3 +-
 packages/platform-bun/src/BunClusterSocket.ts |   3 +-
 packages/platform-node/src/NodeClusterHttp.ts |   3 +-
 .../platform-node/src/NodeClusterSocket.ts    |   3 +-
 .../test/cluster/MessageStorageTest.ts        |   7 +
 .../test/cluster/SqlMessageStorage.test.ts    |  86 +++++++++++-
 10 files changed, 231 insertions(+), 27 deletions(-)
 create mode 100644 .changeset/hash-sql-message-dedupe-keys.md

diff --git a/.changeset/hash-sql-message-dedupe-keys.md b/.changeset/hash-sql-message-dedupe-keys.md
new file mode 100644
index 00000000000..34d1eb63c58
--- /dev/null
+++ b/.changeset/hash-sql-message-dedupe-keys.md
@@ -0,0 +1,13 @@
+---
+"effect": patch
+"@effect/platform-node": patch
+"@effect/platform-bun": patch
+---
+
+unstable/cluster: hash 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, so keys of any length work on PostgreSQL, MySQL, MSSQL, and SQLite. A migration adds a nullable, non-unique plaintext `primary_key` column for diagnostics.
+
+Rows written before this change store the plaintext key in `message_id`; save-time deduplication and `requestIdForPrimaryKey` fall back to a plaintext read for those rows, so in-flight requests keep deduplicating across the upgrade.
+
+`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.
diff --git a/packages/effect/src/unstable/cluster/Envelope.ts b/packages/effect/src/unstable/cluster/Envelope.ts
index 558ac98561f..6bc3617a7f6 100644
--- a/packages/effect/src/unstable/cluster/Envelope.ts
+++ b/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}`
diff --git a/packages/effect/src/unstable/cluster/SingleRunner.ts b/packages/effect/src/unstable/cluster/SingleRunner.ts
index 3b150c9a733..bfa9e43106b 100644
--- a/packages/effect/src/unstable/cluster/SingleRunner.ts
+++ b/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 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),
diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
index 63c1c38c480..e6de9fa810b 100644
--- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
+++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
@@ -8,13 +8,20 @@
  * storage constructor, layers, migrations, optional table prefixes, and the row
  * mapping needed by encoded message storage.
  *
+ * Request deduplication keys are hashed with the `Crypto` service before they
+ * are written to the unique `message_id` column, so composed keys of any
+ * length are supported; the plaintext key is stored in the non-unique
+ * `primary_key` column for diagnostics.
+ *
  * @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 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 +68,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,9 +92,35 @@ 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.
+  // `message_id` therefore stores a SHA-256 digest of the composed key: 256
+  // bits keeps the collision probability negligible (birthday bound 2^128),
+  // and the 64-character hex encoding fits every dialect's indexable column
+  // width. The digest never contains "/" while legacy plaintext keys always
+  // do, so the two encodings cannot collide and legacy rows can be matched
+  // with a dual-read fallback. The plaintext is kept in the non-unique
+  // `primary_key` column for diagnostics.
+  const encoder = new TextEncoder()
+  const hashPrimaryKey = (primaryKey: string): Effect.Effect<string, PlatformError.PlatformError> =>
+    Effect.map(crypto.digest("SHA-256", encoder.encode(primaryKey)), (bytes) => {
+      let hex = ""
+      for (let i = 0; i < bytes.length; i++) {
+        hex += bytes[i].toString(16).padStart(2, "0")
+      }
+      return hex
+    })
+
+  // Rows written before `message_id` was hashed store the plaintext composed
+  // key. Keys longer than the column's 255-character limit can never exist as
+  // legacy rows, so the fallback read is skipped for them.
+  const mayHaveLegacyRow = (primaryKey: string): boolean => primaryKey.length <= 255
+
   const envelopeToRow = (
     envelope: Envelope.Encoded,
     message_id: string | null,
+    primary_key: string | null,
     deliver_at: number | null
   ): MessageRow => {
     switch (envelope._tag) {
@@ -94,6 +128,7 @@ export const make: (options?: {
         return {
           id: envelope.requestId,
           message_id,
+          primary_key,
           shard_id: ShardId.toString(envelope.address.shardId),
           entity_type: envelope.address.entityType,
           entity_id: envelope.address.entityId,
@@ -118,6 +153,7 @@ export const make: (options?: {
         return {
           id: envelope.id,
           message_id,
+          primary_key,
           shard_id: ShardId.toString(envelope.address.shardId),
           entity_type: envelope.address.entityType,
           entity_id: envelope.address.entityId,
@@ -136,6 +172,7 @@ export const make: (options?: {
         return {
           id: envelope.id,
           message_id,
+          primary_key,
           shard_id: ShardId.toString(envelope.address.shardId),
           entity_type: envelope.address.entityType,
           entity_id: envelope.address.entityId,
@@ -238,6 +275,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
@@ -406,18 +451,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(hashPrimaryKey(primaryKey), (messageId) => {
+            const row = envelopeToRow(envelope, messageId, primaryKey, 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, 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 +545,15 @@ export const make: (options?: {
     ),
 
     requestIdForPrimaryKey: (primaryKey) =>
-      sql<{ id: string | bigint }>`SELECT id FROM ${messagesTableSql} WHERE message_id = ${primaryKey}`.pipe(
+      hashPrimaryKey(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 +710,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 before they are stored in the fixed-width
+ * `message_id` column.
  *
  * **Gotchas**
  *
@@ -662,7 +729,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 +742,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)
   )
@@ -977,6 +1044,35 @@ const migrations = (options?: {
           // sqlite
           Effect.void
       })
+    }),
+    "0003_message_id_digest": Effect.gen(function*() {
+      const sql = (yield* SqlClient.SqlClient).withoutTransforms()
+      const messagesTableSql = sql(messagesTable)
+
+      // `message_id` now stores a SHA-256 digest of the composed primary key.
+      // Add a nullable, non-unique plaintext `primary_key` column for
+      // diagnostics; it is never indexed, so it cannot hit dialect
+      // index-length limits.
+      yield* sql.onDialectOrElse({
+        mssql: () =>
+          sql`
+            IF COL_LENGTH(N'${messagesTableSql}', 'primary_key') IS NULL
+            ALTER TABLE ${messagesTableSql} ADD primary_key VARCHAR(MAX);
+          `,
+        mysql: () =>
+          sql`
+            ALTER TABLE ${messagesTableSql} ADD COLUMN primary_key TEXT;
+          `.unprepared.pipe(Effect.ignore),
+        pg: () =>
+          sql`
+            ALTER TABLE ${messagesTableSql} ADD COLUMN IF NOT EXISTS primary_key TEXT;
+          `,
+        orElse: () =>
+          // sqlite
+          sql`
+            ALTER TABLE ${messagesTableSql} ADD COLUMN primary_key TEXT;
+          `
+      })
     })
   })
 }
@@ -1011,6 +1107,7 @@ const replyFromRow = (row: ReplyRow): Reply.Encoded =>
 type MessageRow = {
   readonly id: string | bigint
   readonly message_id: string | null
+  readonly primary_key: string | null
   readonly shard_id: string
   readonly entity_type: string
   readonly entity_id: string
diff --git a/packages/platform-bun/src/BunClusterHttp.ts b/packages/platform-bun/src/BunClusterHttp.ts
index ca727c8e895..9eeaa32fce9 100644
--- a/packages/platform-bun/src/BunClusterHttp.ts
+++ b/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"
diff --git a/packages/platform-bun/src/BunClusterSocket.ts b/packages/platform-bun/src/BunClusterSocket.ts
index 22f3bfa30d6..a59a2f553b2 100644
--- a/packages/platform-bun/src/BunClusterSocket.ts
+++ b/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"
diff --git a/packages/platform-node/src/NodeClusterHttp.ts b/packages/platform-node/src/NodeClusterHttp.ts
index 82543c8061b..d66954ef321 100644
--- a/packages/platform-node/src/NodeClusterHttp.ts
+++ b/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"
diff --git a/packages/platform-node/src/NodeClusterSocket.ts b/packages/platform-node/src/NodeClusterSocket.ts
index cd8f1d97629..002479c358b 100644
--- a/packages/platform-node/src/NodeClusterSocket.ts
+++ b/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"
diff --git a/packages/platform-node/test/cluster/MessageStorageTest.ts b/packages/platform-node/test/cluster/MessageStorageTest.ts
index f580519bc78..972719fda0f 100644
--- a/packages/platform-node/test/cluster/MessageStorageTest.ts
+++ b/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: {
diff --git a/packages/platform-node/test/cluster/SqlMessageStorage.test.ts b/packages/platform-node/test/cluster/SqlMessageStorage.test.ts
index 16deaaa863d..996c62d9e25 100644
--- a/packages/platform-node/test/cluster/SqlMessageStorage.test.ts
+++ b/packages/platform-node/test/cluster/SqlMessageStorage.test.ts
@@ -1,13 +1,21 @@
-import { NodeFileSystem } from "@effect/platform-node"
+import { NodeCrypto, NodeFileSystem } from "@effect/platform-node"
 import { SqliteClient } from "@effect/sql-sqlite-node"
 import { assert, describe, expect, it } from "@effect/vitest"
 import { Effect, Fiber, FileSystem, Latch, Layer, Option } from "effect"
 import { TestClock } from "effect/testing"
-import { Message, MessageStorage, ShardingConfig, Snowflake, SqlMessageStorage } from "effect/unstable/cluster"
+import {
+  Envelope,
+  Message,
+  MessageStorage,
+  ShardingConfig,
+  Snowflake,
+  SqlMessageStorage
+} from "effect/unstable/cluster"
 import { SqlClient } from "effect/unstable/sql"
 import { MysqlContainer } from "../fixtures/mysql2-utils.ts"
 import { PgContainer } from "../fixtures/pg-utils.ts"
 import {
+  LongKeyRpc,
   makeAckChunk,
   makeChunkReply,
   makeReply,
@@ -18,7 +26,7 @@ import {
 
 const StorageLive = SqlMessageStorage.layer.pipe(
   Layer.provideMerge(Snowflake.layerGenerator),
-  Layer.provide(ShardingConfig.layerDefaults)
+  Layer.provide([ShardingConfig.layerDefaults, NodeCrypto.layer])
 )
 
 const truncate = Effect.gen(function*() {
@@ -135,6 +143,78 @@ describe("SqlMessageStorage", () => {
           expect(result._tag).toEqual("Duplicate")
         }))
 
+      it.effect("hashes primary keys longer than the message_id column", () =>
+        Effect.gen(function*() {
+          yield* truncate
+
+          const storage = yield* MessageStorage.MessageStorage
+          const longId = "long-key-".repeat(50)
+          const request = yield* makeRequest({
+            rpc: LongKeyRpc,
+            payload: LongKeyRpc.payloadSchema.make({ id: longId })
+          })
+          const result = yield* storage.saveRequest(request)
+          expect(result._tag).toEqual("Success")
+
+          const duplicate = yield* storage.saveRequest(
+            yield* makeRequest({
+              rpc: LongKeyRpc,
+              payload: LongKeyRpc.payloadSchema.make({ id: longId })
+            })
+          )
+          expect(duplicate._tag).toEqual("Duplicate")
+
+          const requestId = yield* storage.requestIdForPrimaryKey({
+            address: request.envelope.address,
+            tag: request.envelope.tag,
+            id: longId
+          })
+          expect(requestId).toEqual(Option.some(request.envelope.requestId))
+
+          const sql = yield* SqlClient.SqlClient
+          const rows = yield* sql<
+            { message_id: string; primary_key: string }
+          >`SELECT message_id, primary_key FROM cluster_messages`
+          expect(rows).toHaveLength(1)
+          expect(rows[0].message_id).toMatch(/^[0-9a-f]{64}$/)
+          expect(rows[0].primary_key).toEqual(Envelope.primaryKey(request.envelope))
+        }))
+
+      it.effect("detects duplicates for legacy plaintext rows", () =>
+        Effect.gen(function*() {
+          yield* truncate
+
+          const sql = yield* SqlClient.SqlClient
+          const storage = yield* MessageStorage.MessageStorage
+          const request = yield* makeRequest({
+            rpc: PrimaryKeyTest,
+            payload: PrimaryKeyTest.payloadSchema.make({ id: 456 })
+          })
+          yield* storage.saveRequest(request)
+
+          // simulate a row written before message_id values were hashed
+          const plaintext = Envelope.primaryKey(request.envelope)!
+          yield* sql`UPDATE cluster_messages SET message_id = ${plaintext} WHERE id = ${
+            String(request.envelope.requestId)
+          }`
+
+          const result = yield* storage.saveRequest(
+            yield* makeRequest({
+              rpc: PrimaryKeyTest,
+              payload: PrimaryKeyTest.payloadSchema.make({ id: 456 })
+            })
+          )
+          assert(result._tag === "Duplicate")
+          expect(result.originalId).toEqual(request.envelope.requestId)
+
+          const requestId = yield* storage.requestIdForPrimaryKey({
+            address: request.envelope.address,
+            tag: request.envelope.tag,
+            id: "456"
+          })
+          expect(requestId).toEqual(Option.some(request.envelope.requestId))
+        }))
+
       it.effect("unprocessedMessages", () =>
         Effect.gen(function*() {
           yield* truncate

From 9d84e92e7c0fa5f20c4581d05ee22d06aa9e46b6 Mon Sep 17 00:00:00 2001
From: Tim Smart <hello@timsmart.co>
Date: Sat, 25 Jul 2026 22:55:40 +1200
Subject: [PATCH 2/5] Make legacy dedup fallback dialect-aware for sqlite TEXT
 message_id

sqlite does not enforce VARCHAR widths, so legacy plaintext rows there can
hold composed keys longer than 255 characters; only skip the fallback read
on the width-enforcing dialects.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .../src/unstable/cluster/SqlMessageStorage.ts | 18 ++++++--
 .../test/cluster/SqlMessageStorage.test.ts    | 42 +++++++++++++++++++
 2 files changed, 57 insertions(+), 3 deletions(-)

diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
index e6de9fa810b..8c3b229fa81 100644
--- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
+++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
@@ -113,9 +113,21 @@ export const make: (options?: {
     })
 
   // Rows written before `message_id` was hashed store the plaintext composed
-  // key. Keys longer than the column's 255-character limit can never exist as
-  // legacy rows, so the fallback read is skipped for them.
-  const mayHaveLegacyRow = (primaryKey: string): boolean => primaryKey.length <= 255
+  // key; the fallback reads below exist only for that transition and can be
+  // removed in a future release. On the width-enforcing dialects keys longer
+  // than the column's 255-character limit can never exist as legacy rows, so
+  // the fallback read is skipped for them — but sqlite declares `message_id`
+  // as TEXT and does not enforce VARCHAR widths, so legacy rows there can
+  // hold keys of any length. (MySQL's `INSERT IGNORE` truncated over-long
+  // legacy keys to a 255-character prefix; those rows were already corrupt
+  // for deduplication purposes pre-digest and are deliberately not matched.)
+  const messageIdEnforcesWidth = sql.onDialectOrElse({
+    mssql: () => true,
+    mysql: () => true,
+    pg: () => true,
+    orElse: () => false
+  })
+  const mayHaveLegacyRow = (primaryKey: string): boolean => !messageIdEnforcesWidth || primaryKey.length <= 255
 
   const envelopeToRow = (
     envelope: Envelope.Encoded,
diff --git a/packages/platform-node/test/cluster/SqlMessageStorage.test.ts b/packages/platform-node/test/cluster/SqlMessageStorage.test.ts
index 996c62d9e25..58c104eb198 100644
--- a/packages/platform-node/test/cluster/SqlMessageStorage.test.ts
+++ b/packages/platform-node/test/cluster/SqlMessageStorage.test.ts
@@ -215,6 +215,48 @@ describe("SqlMessageStorage", () => {
           expect(requestId).toEqual(Option.some(request.envelope.requestId))
         }))
 
+      if (label === "sqlite") {
+        // sqlite's TEXT message_id column stored over-long plaintext keys
+        // before hashing, so the legacy fallback must also cover keys longer
+        // than the 255-character limit of the width-enforcing dialects
+        it.effect("detects duplicates for legacy long-key plaintext rows", () =>
+          Effect.gen(function*() {
+            yield* truncate
+
+            const sql = yield* SqlClient.SqlClient
+            const storage = yield* MessageStorage.MessageStorage
+            const longId = "legacy-long-key-".repeat(30)
+            const request = yield* makeRequest({
+              rpc: LongKeyRpc,
+              payload: LongKeyRpc.payloadSchema.make({ id: longId })
+            })
+            yield* storage.saveRequest(request)
+
+            // simulate a row written before message_id values were hashed
+            const plaintext = Envelope.primaryKey(request.envelope)!
+            expect(plaintext.length).toBeGreaterThan(255)
+            yield* sql`UPDATE cluster_messages SET message_id = ${plaintext} WHERE id = ${
+              String(request.envelope.requestId)
+            }`
+
+            const result = yield* storage.saveRequest(
+              yield* makeRequest({
+                rpc: LongKeyRpc,
+                payload: LongKeyRpc.payloadSchema.make({ id: longId })
+              })
+            )
+            assert(result._tag === "Duplicate")
+            expect(result.originalId).toEqual(request.envelope.requestId)
+
+            const requestId = yield* storage.requestIdForPrimaryKey({
+              address: request.envelope.address,
+              tag: request.envelope.tag,
+              id: longId
+            })
+            expect(requestId).toEqual(Option.some(request.envelope.requestId))
+          }))
+      }
+
       it.effect("unprocessedMessages", () =>
         Effect.gen(function*() {
           yield* truncate

From 9ad793bdda804d653b09106ce7aaf58dafb083c3 Mon Sep 17 00:00:00 2001
From: Tim Smart <hello@timsmart.co>
Date: Sat, 25 Jul 2026 23:09:00 +1200
Subject: [PATCH 3/5] Use Encoding.encodeHex for digest encoding, reuse
 selectByMessageId

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .../src/unstable/cluster/SqlMessageStorage.ts | 30 ++++---------------
 1 file changed, 5 insertions(+), 25 deletions(-)

diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
index 8c3b229fa81..dee407b6780 100644
--- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
+++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
@@ -19,6 +19,7 @@
 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"
@@ -104,13 +105,7 @@ export const make: (options?: {
   // `primary_key` column for diagnostics.
   const encoder = new TextEncoder()
   const hashPrimaryKey = (primaryKey: string): Effect.Effect<string, PlatformError.PlatformError> =>
-    Effect.map(crypto.digest("SHA-256", encoder.encode(primaryKey)), (bytes) => {
-      let hex = ""
-      for (let i = 0; i < bytes.length; i++) {
-        hex += bytes[i].toString(16).padStart(2, "0")
-      }
-      return hex
-    })
+    Effect.map(crypto.digest("SHA-256", encoder.encode(primaryKey)), Encoding.encodeHex)
 
   // Rows written before `message_id` was hashed store the plaintext composed
   // key; the fallback reads below exist only for that transition and can be
@@ -307,12 +302,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(
@@ -321,12 +311,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) =>
@@ -368,12 +353,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 })

From 2320bfd67e79fd96dcf566a6878e65704fe655bc Mon Sep 17 00:00:00 2001
From: Tim Smart <hello@timsmart.co>
Date: Sat, 25 Jul 2026 23:15:34 +1200
Subject: [PATCH 4/5] Drop plaintext primary_key diagnostic column and 0003
 migration

The composed key's components are already readable via the entity_type,
entity_id, tag, and payload columns, so the PR is now a pure code change
with no schema change.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .changeset/hash-sql-message-dedupe-keys.md    |  2 +-
 .../src/unstable/cluster/SqlMessageStorage.ts | 46 +++----------------
 .../test/cluster/SqlMessageStorage.test.ts    |  5 +-
 3 files changed, 8 insertions(+), 45 deletions(-)

diff --git a/.changeset/hash-sql-message-dedupe-keys.md b/.changeset/hash-sql-message-dedupe-keys.md
index 34d1eb63c58..a878949e5d6 100644
--- a/.changeset/hash-sql-message-dedupe-keys.md
+++ b/.changeset/hash-sql-message-dedupe-keys.md
@@ -6,7 +6,7 @@
 
 unstable/cluster: hash 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, so keys of any length work on PostgreSQL, MySQL, MSSQL, and SQLite. A migration adds a nullable, non-unique plaintext `primary_key` column for diagnostics.
+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, so keys of any length work on PostgreSQL, MySQL, MSSQL, and SQLite. No schema change is required; the key's components remain readable via the `entity_type`, `entity_id`, `tag`, and `payload` columns.
 
 Rows written before this change store the plaintext key in `message_id`; save-time deduplication and `requestIdForPrimaryKey` fall back to a plaintext read for those rows, so in-flight requests keep deduplicating across the upgrade.
 
diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
index dee407b6780..7eb8644859a 100644
--- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
+++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
@@ -10,8 +10,8 @@
  *
  * Request deduplication keys are hashed with the `Crypto` service before they
  * are written to the unique `message_id` column, so composed keys of any
- * length are supported; the plaintext key is stored in the non-unique
- * `primary_key` column for diagnostics.
+ * length are supported. The key's components remain readable via the
+ * `entity_type`, `entity_id`, `tag`, and `payload` columns.
  *
  * @since 4.0.0
  */
@@ -101,8 +101,8 @@ export const make: (options?: {
   // and the 64-character hex encoding fits every dialect's indexable column
   // width. The digest never contains "/" while legacy plaintext keys always
   // do, so the two encodings cannot collide and legacy rows can be matched
-  // with a dual-read fallback. The plaintext is kept in the non-unique
-  // `primary_key` column for diagnostics.
+  // with a dual-read fallback. The key's components remain readable via the
+  // entity_type, entity_id, tag, and payload columns.
   const encoder = new TextEncoder()
   const hashPrimaryKey = (primaryKey: string): Effect.Effect<string, PlatformError.PlatformError> =>
     Effect.map(crypto.digest("SHA-256", encoder.encode(primaryKey)), Encoding.encodeHex)
@@ -127,7 +127,6 @@ export const make: (options?: {
   const envelopeToRow = (
     envelope: Envelope.Encoded,
     message_id: string | null,
-    primary_key: string | null,
     deliver_at: number | null
   ): MessageRow => {
     switch (envelope._tag) {
@@ -135,7 +134,6 @@ export const make: (options?: {
         return {
           id: envelope.requestId,
           message_id,
-          primary_key,
           shard_id: ShardId.toString(envelope.address.shardId),
           entity_type: envelope.address.entityType,
           entity_id: envelope.address.entityId,
@@ -160,7 +158,6 @@ export const make: (options?: {
         return {
           id: envelope.id,
           message_id,
-          primary_key,
           shard_id: ShardId.toString(envelope.address.shardId),
           entity_type: envelope.address.entityType,
           entity_id: envelope.address.entityId,
@@ -179,7 +176,6 @@ export const make: (options?: {
         return {
           id: envelope.id,
           message_id,
-          primary_key,
           shard_id: ShardId.toString(envelope.address.shardId),
           entity_type: envelope.address.entityType,
           entity_id: envelope.address.entityId,
@@ -446,7 +442,7 @@ export const make: (options?: {
         let insert: Effect.Effect<ReadonlyArray<Row>, SqlError | PlatformError.PlatformError>
         if (primaryKey !== null) {
           insert = Effect.flatMap(hashPrimaryKey(primaryKey), (messageId) => {
-            const row = envelopeToRow(envelope, messageId, primaryKey, deliverAt)
+            const row = envelopeToRow(envelope, messageId, deliverAt)
             if (!mayHaveLegacyRow(primaryKey)) {
               return insertEnvelope(row, messageId)
             }
@@ -456,7 +452,7 @@ export const make: (options?: {
             )
           })
         } else {
-          const row = envelopeToRow(envelope, null, null, deliverAt)
+          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(
@@ -1036,35 +1032,6 @@ const migrations = (options?: {
           // sqlite
           Effect.void
       })
-    }),
-    "0003_message_id_digest": Effect.gen(function*() {
-      const sql = (yield* SqlClient.SqlClient).withoutTransforms()
-      const messagesTableSql = sql(messagesTable)
-
-      // `message_id` now stores a SHA-256 digest of the composed primary key.
-      // Add a nullable, non-unique plaintext `primary_key` column for
-      // diagnostics; it is never indexed, so it cannot hit dialect
-      // index-length limits.
-      yield* sql.onDialectOrElse({
-        mssql: () =>
-          sql`
-            IF COL_LENGTH(N'${messagesTableSql}', 'primary_key') IS NULL
-            ALTER TABLE ${messagesTableSql} ADD primary_key VARCHAR(MAX);
-          `,
-        mysql: () =>
-          sql`
-            ALTER TABLE ${messagesTableSql} ADD COLUMN primary_key TEXT;
-          `.unprepared.pipe(Effect.ignore),
-        pg: () =>
-          sql`
-            ALTER TABLE ${messagesTableSql} ADD COLUMN IF NOT EXISTS primary_key TEXT;
-          `,
-        orElse: () =>
-          // sqlite
-          sql`
-            ALTER TABLE ${messagesTableSql} ADD COLUMN primary_key TEXT;
-          `
-      })
     })
   })
 }
@@ -1099,7 +1066,6 @@ const replyFromRow = (row: ReplyRow): Reply.Encoded =>
 type MessageRow = {
   readonly id: string | bigint
   readonly message_id: string | null
-  readonly primary_key: string | null
   readonly shard_id: string
   readonly entity_type: string
   readonly entity_id: string
diff --git a/packages/platform-node/test/cluster/SqlMessageStorage.test.ts b/packages/platform-node/test/cluster/SqlMessageStorage.test.ts
index 58c104eb198..cfa7afc8bc1 100644
--- a/packages/platform-node/test/cluster/SqlMessageStorage.test.ts
+++ b/packages/platform-node/test/cluster/SqlMessageStorage.test.ts
@@ -172,12 +172,9 @@ describe("SqlMessageStorage", () => {
           expect(requestId).toEqual(Option.some(request.envelope.requestId))
 
           const sql = yield* SqlClient.SqlClient
-          const rows = yield* sql<
-            { message_id: string; primary_key: string }
-          >`SELECT message_id, primary_key FROM cluster_messages`
+          const rows = yield* sql<{ message_id: string }>`SELECT message_id FROM cluster_messages`
           expect(rows).toHaveLength(1)
           expect(rows[0].message_id).toMatch(/^[0-9a-f]{64}$/)
-          expect(rows[0].primary_key).toEqual(Envelope.primaryKey(request.envelope))
         }))
 
       it.effect("detects duplicates for legacy plaintext rows", () =>

From a1f96bcae39e22dc4959bb611415770569d0ec7e Mon Sep 17 00:00:00 2001
From: Tim Smart <hello@timsmart.co>
Date: Sat, 25 Jul 2026 23:24:58 +1200
Subject: [PATCH 5/5] Only hash primary keys that exceed the message_id column
 width

Keys within 255 characters stay byte-identical plaintext, keeping
deduplication compatible with rows written by previous versions with no
fallback reads. The plaintext fallback now covers only over-length legacy
keys on sqlite, whose TEXT column stored them before digests existed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .changeset/hash-sql-message-dedupe-keys.md    |  6 +--
 .../src/unstable/cluster/SingleRunner.ts      |  2 +-
 .../src/unstable/cluster/SqlMessageStorage.ts | 45 ++++++++-----------
 .../test/cluster/SqlMessageStorage.test.ts    | 11 ++---
 4 files changed, 28 insertions(+), 36 deletions(-)

diff --git a/.changeset/hash-sql-message-dedupe-keys.md b/.changeset/hash-sql-message-dedupe-keys.md
index a878949e5d6..9e386e0193c 100644
--- a/.changeset/hash-sql-message-dedupe-keys.md
+++ b/.changeset/hash-sql-message-dedupe-keys.md
@@ -4,10 +4,8 @@
 "@effect/platform-bun": patch
 ---
 
-unstable/cluster: hash SQL message deduplication keys to prevent `message_id` overflow, closes #6317.
+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, so keys of any length work on PostgreSQL, MySQL, MSSQL, and SQLite. No schema change is required; the key's components remain readable via the `entity_type`, `entity_id`, `tag`, and `payload` columns.
-
-Rows written before this change store the plaintext key in `message_id`; save-time deduplication and `requestIdForPrimaryKey` fall back to a plaintext read for those rows, so in-flight requests keep deduplicating across the upgrade.
+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.
diff --git a/packages/effect/src/unstable/cluster/SingleRunner.ts b/packages/effect/src/unstable/cluster/SingleRunner.ts
index bfa9e43106b..3f3cdb90690 100644
--- a/packages/effect/src/unstable/cluster/SingleRunner.ts
+++ b/packages/effect/src/unstable/cluster/SingleRunner.ts
@@ -44,7 +44,7 @@ import * as SqlRunnerStorage from "./SqlRunnerStorage.ts"
  *
  * - Even when `runnerStorage` is `"memory"`, message storage remains
  *   SQL-backed, so callers must still provide `SqlClient` and `Crypto.Crypto`
- *   (used to hash message deduplication keys).
+ *   (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.
  *
diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
index 7eb8644859a..c63c403675e 100644
--- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
+++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
@@ -8,10 +8,10 @@
  * storage constructor, layers, migrations, optional table prefixes, and the row
  * mapping needed by encoded message storage.
  *
- * Request deduplication keys are hashed with the `Crypto` service before they
- * are written to the unique `message_id` column, so composed keys of any
- * length are supported. The key's components remain readable via the
- * `entity_type`, `entity_id`, `tag`, and `payload` columns.
+ * 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
  */
@@ -96,33 +96,26 @@ export const make: (options?: {
   // 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.
-  // `message_id` therefore stores a SHA-256 digest of the composed key: 256
-  // bits keeps the collision probability negligible (birthday bound 2^128),
-  // and the 64-character hex encoding fits every dialect's indexable column
-  // width. The digest never contains "/" while legacy plaintext keys always
-  // do, so the two encodings cannot collide and legacy rows can be matched
-  // with a dual-read fallback. The key's components remain readable via the
-  // entity_type, entity_id, tag, and payload columns.
+  // 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 hashPrimaryKey = (primaryKey: string): Effect.Effect<string, PlatformError.PlatformError> =>
-    Effect.map(crypto.digest("SHA-256", encoder.encode(primaryKey)), Encoding.encodeHex)
+  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)
 
-  // Rows written before `message_id` was hashed store the plaintext composed
-  // key; the fallback reads below exist only for that transition and can be
-  // removed in a future release. On the width-enforcing dialects keys longer
-  // than the column's 255-character limit can never exist as legacy rows, so
-  // the fallback read is skipped for them — but sqlite declares `message_id`
-  // as TEXT and does not enforce VARCHAR widths, so legacy rows there can
-  // hold keys of any length. (MySQL's `INSERT IGNORE` truncated over-long
-  // legacy keys to a 255-character prefix; those rows were already corrupt
-  // for deduplication purposes pre-digest and are deliberately not matched.)
   const messageIdEnforcesWidth = sql.onDialectOrElse({
     mssql: () => true,
     mysql: () => true,
     pg: () => true,
     orElse: () => false
   })
-  const mayHaveLegacyRow = (primaryKey: string): boolean => !messageIdEnforcesWidth || primaryKey.length <= 255
+  // 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,
@@ -441,7 +434,7 @@ export const make: (options?: {
       Effect.suspend(() => {
         let insert: Effect.Effect<ReadonlyArray<Row>, SqlError | PlatformError.PlatformError>
         if (primaryKey !== null) {
-          insert = Effect.flatMap(hashPrimaryKey(primaryKey), (messageId) => {
+          insert = Effect.flatMap(messageIdForPrimaryKey(primaryKey), (messageId) => {
             const row = envelopeToRow(envelope, messageId, deliverAt)
             if (!mayHaveLegacyRow(primaryKey)) {
               return insertEnvelope(row, messageId)
@@ -533,7 +526,7 @@ export const make: (options?: {
     ),
 
     requestIdForPrimaryKey: (primaryKey) =>
-      hashPrimaryKey(primaryKey).pipe(
+      messageIdForPrimaryKey(primaryKey).pipe(
         Effect.flatMap((messageId) =>
           sql<{ id: string | bigint }>`SELECT id FROM ${messagesTableSql} WHERE message_id = ${messageId}`
         ),
@@ -699,7 +692,7 @@ export const make: (options?: {
  * The layer runs the SQL migrations through `make`, provides `MessageStorage`,
  * and supplies `Snowflake.layerGenerator` internally. Callers still provide
  * `SqlClient`, `ShardingConfig`, and `Crypto.Crypto`, which is used to hash
- * message deduplication keys before they are stored in the fixed-width
+ * message deduplication keys that would overflow the fixed-width
  * `message_id` column.
  *
  * **Gotchas**
diff --git a/packages/platform-node/test/cluster/SqlMessageStorage.test.ts b/packages/platform-node/test/cluster/SqlMessageStorage.test.ts
index cfa7afc8bc1..266895edd2e 100644
--- a/packages/platform-node/test/cluster/SqlMessageStorage.test.ts
+++ b/packages/platform-node/test/cluster/SqlMessageStorage.test.ts
@@ -177,7 +177,7 @@ describe("SqlMessageStorage", () => {
           expect(rows[0].message_id).toMatch(/^[0-9a-f]{64}$/)
         }))
 
-      it.effect("detects duplicates for legacy plaintext rows", () =>
+      it.effect("keeps primary keys within the column width as plaintext", () =>
         Effect.gen(function*() {
           yield* truncate
 
@@ -189,11 +189,12 @@ describe("SqlMessageStorage", () => {
           })
           yield* storage.saveRequest(request)
 
-          // simulate a row written before message_id values were hashed
+          // rows written by previous versions store the plaintext composed
+          // key, so short keys must stay byte-identical to keep deduplicating
           const plaintext = Envelope.primaryKey(request.envelope)!
-          yield* sql`UPDATE cluster_messages SET message_id = ${plaintext} WHERE id = ${
-            String(request.envelope.requestId)
-          }`
+          const rows = yield* sql<{ message_id: string }>`SELECT message_id FROM cluster_messages`
+          expect(rows).toHaveLength(1)
+          expect(rows[0].message_id).toEqual(plaintext)
 
           const result = yield* storage.saveRequest(
             yield* makeRequest({