diff --git a/.changeset/hash-sql-message-dedupe-keys.md b/.changeset/hash-sql-message-dedupe-keys.md new file mode 100644 index 00000000000..9e386e0193c --- /dev/null +++ b/.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. 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..3f3cdb90690 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 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), diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts index 63c1c38c480..c63c403675e 100644 --- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts +++ b/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 => + 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, 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, 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 => +}): Layer.Layer => Layer.effect(MessageStorage.MessageStorage, make(options)).pipe( Layer.provide(Snowflake.layerGenerator) ) 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..266895edd2e 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,118 @@ 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 }>`SELECT message_id FROM cluster_messages` + expect(rows).toHaveLength(1) + expect(rows[0].message_id).toMatch(/^[0-9a-f]{64}$/) + })) + + it.effect("keeps primary keys within the column width as plaintext", () => + 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) + + // 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)! + 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({ + 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)) + })) + + 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