From e19f9e5c2f656274969f648865957327e02ab2e0 Mon Sep 17 00:00:00 2001 From: Stephen Brian King <3913213+sbking@users.noreply.github.com> Date: Thu, 16 Jul 2026 16:26:01 -0600 Subject: [PATCH 1/4] Isolate SqlMessageStorage tests with per-test table prefixes --- .../test/cluster/SqlMessageStorage.test.ts | 45 +++++++------------ 1 file changed, 16 insertions(+), 29 deletions(-) diff --git a/packages/platform-node/test/cluster/SqlMessageStorage.test.ts b/packages/platform-node/test/cluster/SqlMessageStorage.test.ts index 16deaaa863d..decdcd408d9 100644 --- a/packages/platform-node/test/cluster/SqlMessageStorage.test.ts +++ b/packages/platform-node/test/cluster/SqlMessageStorage.test.ts @@ -16,16 +16,13 @@ import { StreamRpc } from "./MessageStorageTest.ts" -const StorageLive = SqlMessageStorage.layer.pipe( - Layer.provideMerge(Snowflake.layerGenerator), - Layer.provide(ShardingConfig.layerDefaults) -) - -const truncate = Effect.gen(function*() { - const sql = yield* SqlClient.SqlClient - yield* sql`DELETE FROM cluster_replies` - yield* sql`DELETE FROM cluster_messages` -}) +// Each test provides its own storage layer with a distinct table prefix, so +// concurrently running tests never share tables. +const storageLive = (prefix: string) => + SqlMessageStorage.layerWith({ prefix }).pipe( + Layer.provideMerge(Snowflake.layerGenerator), + Layer.provide(ShardingConfig.layerDefaults) + ) describe("SqlMessageStorage", () => { ;([ @@ -33,7 +30,7 @@ describe("SqlMessageStorage", () => { ["mysql", Layer.orDie(MysqlContainer.layerClient)], ["sqlite", Layer.orDie(SqliteLayer)] ] as const).forEach(([label, layer]) => { - it.layer(StorageLive.pipe(Layer.provideMerge(layer)), { + it.layer(layer, { timeout: 120000 })(label, (it) => { it.effect("saveRequest", () => @@ -59,7 +56,7 @@ describe("SqlMessageStorage", () => { messages = yield* storage.unprocessedMessages([request.envelope.address.shardId]) expect(messages).toHaveLength(5) expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([6, 7, 8, 9, 10]) - })) + }).pipe(Effect.provide(storageLive("save_request")))) it.effect("saveReply + saveRequest duplicate", () => Effect.gen(function*() { @@ -113,12 +110,10 @@ describe("SqlMessageStorage", () => { } const error = yield* Effect.flip(Fiber.join(fiber)) expect(error._tag).toEqual("PersistenceError") - })) + }).pipe(Effect.provide(storageLive("stream_replies")))) it.effect("detects duplicates", () => Effect.gen(function*() { - yield* truncate - const storage = yield* MessageStorage.MessageStorage yield* storage.saveRequest( yield* makeRequest({ @@ -133,12 +128,10 @@ describe("SqlMessageStorage", () => { }) ) expect(result._tag).toEqual("Duplicate") - })) + }).pipe(Effect.provide(storageLive("detects_duplicates")))) it.effect("unprocessedMessages", () => Effect.gen(function*() { - yield* truncate - const storage = yield* MessageStorage.MessageStorage const request = yield* makeRequest() yield* storage.saveRequest(request) @@ -149,24 +142,20 @@ describe("SqlMessageStorage", () => { yield* storage.saveRequest(yield* makeRequest()) messages = yield* storage.unprocessedMessages([request.envelope.address.shardId]) expect(messages).toHaveLength(1) - })) + }).pipe(Effect.provide(storageLive("unprocessed_messages")))) it.effect("unprocessedMessages excludes complete requests", () => Effect.gen(function*() { - yield* truncate - const storage = yield* MessageStorage.MessageStorage const request = yield* makeRequest() yield* storage.saveRequest(request) yield* storage.saveReply(yield* makeReply(request)) const messages = yield* storage.unprocessedMessages([request.envelope.address.shardId]) expect(messages).toHaveLength(0) - })) + }).pipe(Effect.provide(storageLive("unprocessed_complete")))) it.effect("repliesFor", () => Effect.gen(function*() { - yield* truncate - const storage = yield* MessageStorage.MessageStorage const request = yield* makeRequest() yield* storage.saveRequest(request) @@ -176,7 +165,7 @@ describe("SqlMessageStorage", () => { replies = yield* storage.repliesFor([request]) expect(replies).toHaveLength(1) expect(replies[0].requestId).toEqual(request.envelope.requestId) - })) + }).pipe(Effect.provide(storageLive("replies_for")))) it.effect("registerReplyHandler", () => Effect.gen(function*() { @@ -194,12 +183,10 @@ describe("SqlMessageStorage", () => { yield* storage.saveReply(yield* makeReply(request)) yield* latch.await yield* Fiber.await(fiber) - })) + }).pipe(Effect.provide(storageLive("register_reply_handler")))) it.effect("unprocessedMessagesById", () => Effect.gen(function*() { - yield* truncate - const storage = yield* MessageStorage.MessageStorage const request = yield* makeRequest() yield* storage.saveRequest(request) @@ -208,7 +195,7 @@ describe("SqlMessageStorage", () => { yield* storage.saveReply(yield* makeReply(request)) messages = yield* storage.unprocessedMessagesById([request.envelope.requestId]) expect(messages).toHaveLength(0) - })) + }).pipe(Effect.provide(storageLive("unprocessed_by_id")))) }) }) }) From 171f5452acef93972aed7824bfdf4445dc53c9cf Mon Sep 17 00:00:00 2001 From: Stephen Brian King <3913213+sbking@users.noreply.github.com> Date: Thu, 16 Jul 2026 10:31:58 -0600 Subject: [PATCH 2/4] Fix invalid FOR UPDATE clause on SQL Server in cluster SqlMessageStorage --- .changeset/fix-cluster-mssql-for-update.md | 5 +++++ packages/effect/src/unstable/cluster/SqlMessageStorage.ts | 2 ++ 2 files changed, 7 insertions(+) create mode 100644 .changeset/fix-cluster-mssql-for-update.md diff --git a/.changeset/fix-cluster-mssql-for-update.md b/.changeset/fix-cluster-mssql-for-update.md new file mode 100644 index 00000000000..c1666da14f2 --- /dev/null +++ b/.changeset/fix-cluster-mssql-for-update.md @@ -0,0 +1,5 @@ +--- +"effect": patch +--- + +Fix cluster SqlMessageStorage emitting an invalid `FOR UPDATE` clause on SQL Server, which caused every unprocessed-message read to fail with a syntax error. diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts index 63c1c38c480..8adff92f67b 100644 --- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts @@ -343,6 +343,8 @@ export const make: (options?: { }) const forUpdate = sql.onDialectOrElse({ sqlite: () => sql.literal(""), + // SQL Server uses table hints instead of a trailing FOR UPDATE clause. + mssql: () => sql.literal(""), orElse: () => sql.literal("FOR UPDATE") }) From a546e0e1bffd3ea0c16ed6f3565dd665b16ca211 Mon Sep 17 00:00:00 2001 From: Stephen Brian King Date: Thu, 16 Jul 2026 14:47:47 -0600 Subject: [PATCH 3/4] Fix SQL Server unique constraints, MERGE output and column types in cluster SqlMessageStorage --- .changeset/fix-cluster-mssql-for-update.md | 11 +- .../src/unstable/cluster/SqlMessageStorage.ts | 140 +++++++++++++----- 2 files changed, 111 insertions(+), 40 deletions(-) diff --git a/.changeset/fix-cluster-mssql-for-update.md b/.changeset/fix-cluster-mssql-for-update.md index c1666da14f2..2d858644706 100644 --- a/.changeset/fix-cluster-mssql-for-update.md +++ b/.changeset/fix-cluster-mssql-for-update.md @@ -2,4 +2,13 @@ "effect": patch --- -Fix cluster SqlMessageStorage emitting an invalid `FOR UPDATE` clause on SQL Server, which caused every unprocessed-message read to fail with a syntax error. +Fix SQL Server support in cluster SqlMessageStorage. + +The mssql code paths contained failures across core storage paths: + +- The unprocessed-message reads emitted a trailing `FOR UPDATE` clause, which T-SQL only allows on cursors, so the reads failed with a syntax error. The mssql dialect now omits the clause (as sqlite does) without adding an `UPDLOCK` hint, so concurrent duplicate observation stays within the existing at-least-once delivery contract. +- The messages and replies tables declared plain `UNIQUE` constraints on `message_id`, `(request_id, kind)` and `(request_id, sequence)`. SQL Server treats NULLs as equal in unique constraints, so the second message without a primary key — and the second chunk reply of any stream — failed with a unique violation. The constraints are replaced with filtered unique indexes (`WHERE ... IS NOT NULL`), matching the NULL-exempt semantics of the other dialects. +- The `insertEnvelope` MERGE statement used subqueries in its `OUTPUT` clause, which SQL Server rejects (error 10705), so saving a request with a primary key always failed — and its duplicate-detection logic was wrong besides, since a MERGE with only `WHEN NOT MATCHED` emits no OUTPUT row for an existing key. +- The `payload` and `headers` columns were declared as the deprecated `TEXT` type. The mssql client binds NULL parameters as `BIT`, which has no implicit conversion to `TEXT`, so inserting envelopes with NULL payloads (e.g. `AckChunk`) failed with an operand type clash. The columns are now `NVARCHAR(MAX)`. + +A new `0003_mssql_schema_fixes` migration converges existing SQL Server databases (a no-op on other dialects): it drops the anonymous and named unique constraints, creates the filtered unique indexes, and alters the `TEXT` columns to `NVARCHAR(MAX)` in place, preserving existing data. diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts index 8adff92f67b..94259ec9964 100644 --- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts @@ -279,37 +279,17 @@ export const make: (options?: { ON target.message_id = source.message_id WHEN NOT MATCHED THEN INSERT ${sql.insert(row)} - OUTPUT - inserted.id, - CASE - WHEN inserted.id IS NULL THEN ( - SELECT r.id, r.kind, r.payload - FROM ${repliesTableSql} r - WHERE r.id = target.last_reply_id - ) - END as reply_id, - CASE - WHEN inserted.id IS NULL THEN ( - SELECT r.kind - FROM ${repliesTableSql} r - WHERE r.id = target.last_reply_id - ) - END as reply_kind, - CASE - WHEN inserted.id IS NULL THEN ( - SELECT r.payload - FROM ${repliesTableSql} r - WHERE r.id = target.last_reply_id - ) - END as reply_payload, - CASE - WHEN inserted.id IS NULL THEN ( - SELECT r.sequence - FROM ${repliesTableSql} r - WHERE r.id = target.last_reply_id - ) - END as reply_sequence; - `, + OUTPUT inserted.id; + `.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} + ` + })), 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 @@ -713,8 +693,8 @@ const migrations = (options?: { entity_id VARCHAR(255) NOT NULL, kind INT NOT NULL, tag VARCHAR(50), - payload TEXT, - headers TEXT, + payload NVARCHAR(MAX), + headers NVARCHAR(MAX), trace_id VARCHAR(32), span_id VARCHAR(16), sampled BIT, @@ -723,8 +703,7 @@ const migrations = (options?: { reply_id BIGINT, last_reply_id BIGINT, last_read DATETIME, - deliver_at BIGINT, - UNIQUE (message_id) + deliver_at BIGINT ) `, mysql: () => @@ -868,11 +847,9 @@ const migrations = (options?: { rowid BIGINT IDENTITY(1,1), kind INT, request_id BIGINT NOT NULL, - payload TEXT NOT NULL, + payload NVARCHAR(MAX) NOT NULL, sequence INT, - acked BIT NOT NULL DEFAULT 0, - CONSTRAINT ${sql(repliesTable + "_one_exit")} UNIQUE (request_id, kind), - CONSTRAINT ${sql(repliesTable + "_sequence")} UNIQUE (request_id, sequence) + acked BIT NOT NULL DEFAULT 0 ) `, mysql: () => @@ -979,6 +956,91 @@ const migrations = (options?: { // sqlite Effect.void }) + }), + "0003_mssql_schema_fixes": Effect.gen(function*() { + const sql = (yield* SqlClient.SqlClient).withoutTransforms() + const messagesTableSql = sql(messagesTable) + const repliesTableSql = sql(repliesTable) + const messageIdUniqueIndex = `${messagesTable}_message_id_idx` + const oneExitUniqueIndex = `${repliesTable}_one_exit` + const sequenceUniqueIndex = `${repliesTable}_sequence` + + // SQL Server unique constraints treat NULLs as equal, unlike the other + // dialects, so the inline constraints created by 0001 reject rows that + // are legal everywhere else: the second message without a primary key + // (message_id NULL) and the second chunk reply of a request (kind + // NULL). Replace them with filtered unique indexes covering only + // non-NULL keys; the sequence key receives the same filtered treatment + // to match the other dialects' nullable-unique semantics. The + // message_id constraint from 0001 was anonymous, so it is looked up by + // shape before being dropped. + // + // 0001 also declared the payload and headers columns as TEXT, which is + // deprecated and has no conversion from BIT, the type the client binds + // NULL parameters as. Convert them to NVARCHAR(MAX). + yield* sql.onDialectOrElse({ + mssql: () => + sql` + DECLARE @constraint_name sysname; + DECLARE @drop_constraint_sql nvarchar(max); + + SELECT @constraint_name = kc.name + FROM sys.key_constraints AS kc + WHERE kc.parent_object_id = OBJECT_ID(N'${messagesTableSql}') + AND kc.type = 'UQ' + AND EXISTS ( + SELECT 1 + FROM sys.index_columns AS ic + INNER JOIN sys.columns AS c + ON c.object_id = ic.object_id + AND c.column_id = ic.column_id + WHERE ic.object_id = kc.parent_object_id + AND ic.index_id = kc.unique_index_id + AND ic.is_included_column = 0 + AND ic.key_ordinal > 0 + GROUP BY ic.object_id, ic.index_id + HAVING COUNT(*) = 1 + AND MAX(CASE WHEN c.name = N'message_id' THEN 1 ELSE 0 END) = 1 + ); + + IF @constraint_name IS NOT NULL + BEGIN + SET @drop_constraint_sql = N'ALTER TABLE ${messagesTableSql} DROP CONSTRAINT ' + QUOTENAME(@constraint_name); + EXEC sp_executesql @drop_constraint_sql; + END; + + IF EXISTS (SELECT * FROM sys.key_constraints WHERE name = ${oneExitUniqueIndex} AND parent_object_id = OBJECT_ID(N'${repliesTableSql}')) + ALTER TABLE ${repliesTableSql} DROP CONSTRAINT ${sql(oneExitUniqueIndex)}; + + IF EXISTS (SELECT * FROM sys.key_constraints WHERE name = ${sequenceUniqueIndex} AND parent_object_id = OBJECT_ID(N'${repliesTableSql}')) + ALTER TABLE ${repliesTableSql} DROP CONSTRAINT ${sql(sequenceUniqueIndex)}; + + IF NOT EXISTS (SELECT * FROM sys.indexes WHERE name = ${messageIdUniqueIndex} AND object_id = OBJECT_ID(N'${messagesTableSql}')) + CREATE UNIQUE INDEX ${sql(messageIdUniqueIndex)} + ON ${messagesTableSql} (message_id) + WHERE message_id IS NOT NULL; + + IF NOT EXISTS (SELECT * FROM sys.indexes WHERE name = ${oneExitUniqueIndex} AND object_id = OBJECT_ID(N'${repliesTableSql}')) + CREATE UNIQUE INDEX ${sql(oneExitUniqueIndex)} + ON ${repliesTableSql} (request_id, kind) + WHERE kind IS NOT NULL; + + IF NOT EXISTS (SELECT * FROM sys.indexes WHERE name = ${sequenceUniqueIndex} AND object_id = OBJECT_ID(N'${repliesTableSql}')) + CREATE UNIQUE INDEX ${sql(sequenceUniqueIndex)} + ON ${repliesTableSql} (request_id, sequence) + WHERE sequence IS NOT NULL; + + IF EXISTS (SELECT * FROM sys.columns WHERE object_id = OBJECT_ID(N'${messagesTableSql}') AND name = N'payload' AND system_type_id = TYPE_ID(N'text')) + ALTER TABLE ${messagesTableSql} ALTER COLUMN payload NVARCHAR(MAX); + + IF EXISTS (SELECT * FROM sys.columns WHERE object_id = OBJECT_ID(N'${messagesTableSql}') AND name = N'headers' AND system_type_id = TYPE_ID(N'text')) + ALTER TABLE ${messagesTableSql} ALTER COLUMN headers NVARCHAR(MAX); + + IF EXISTS (SELECT * FROM sys.columns WHERE object_id = OBJECT_ID(N'${repliesTableSql}') AND name = N'payload' AND system_type_id = TYPE_ID(N'text')) + ALTER TABLE ${repliesTableSql} ALTER COLUMN payload NVARCHAR(MAX) NOT NULL; + `, + orElse: () => Effect.void + }) }) }) } From 9ad877d6ae6ec47b61303de896d326d24b235fc2 Mon Sep 17 00:00:00 2001 From: Stephen Brian King Date: Thu, 16 Jul 2026 14:47:47 -0600 Subject: [PATCH 4/4] Add SQL Server testcontainer coverage for cluster SqlMessageStorage --- packages/platform-node/package.json | 1 + .../test/cluster/SqlMessageStorage.test.ts | 4 +- .../cluster/SqlMessageStorageMssql.test.ts | 237 ++++++++++++++++++ .../test/fixtures/mssql-utils.ts | 41 +++ pnpm-lock.yaml | 15 ++ 5 files changed, 297 insertions(+), 1 deletion(-) create mode 100644 packages/platform-node/test/cluster/SqlMessageStorageMssql.test.ts create mode 100644 packages/platform-node/test/fixtures/mssql-utils.ts diff --git a/packages/platform-node/package.json b/packages/platform-node/package.json index 9c971e7ec27..441200f035e 100644 --- a/packages/platform-node/package.json +++ b/packages/platform-node/package.json @@ -72,6 +72,7 @@ "ioredis": "^5.7.0" }, "devDependencies": { + "@testcontainers/mssqlserver": "^11.14.0", "@testcontainers/mysql": "^11.14.0", "@testcontainers/postgresql": "^11.14.0", "@testcontainers/redis": "^11.14.0", diff --git a/packages/platform-node/test/cluster/SqlMessageStorage.test.ts b/packages/platform-node/test/cluster/SqlMessageStorage.test.ts index decdcd408d9..fcd56ae9531 100644 --- a/packages/platform-node/test/cluster/SqlMessageStorage.test.ts +++ b/packages/platform-node/test/cluster/SqlMessageStorage.test.ts @@ -5,6 +5,7 @@ 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 { SqlClient } from "effect/unstable/sql" +import { MssqlContainer } from "../fixtures/mssql-utils.ts" import { MysqlContainer } from "../fixtures/mysql2-utils.ts" import { PgContainer } from "../fixtures/pg-utils.ts" import { @@ -24,10 +25,11 @@ const storageLive = (prefix: string) => Layer.provide(ShardingConfig.layerDefaults) ) -describe("SqlMessageStorage", () => { +describe("SqlMessageStorage", { timeout: 30_000 }, () => { ;([ ["pg", Layer.orDie(PgContainer.layerClient)], ["mysql", Layer.orDie(MysqlContainer.layerClient)], + ["mssql", Layer.orDie(MssqlContainer.layerClient)], ["sqlite", Layer.orDie(SqliteLayer)] ] as const).forEach(([label, layer]) => { it.layer(layer, { diff --git a/packages/platform-node/test/cluster/SqlMessageStorageMssql.test.ts b/packages/platform-node/test/cluster/SqlMessageStorageMssql.test.ts new file mode 100644 index 00000000000..d7a179683af --- /dev/null +++ b/packages/platform-node/test/cluster/SqlMessageStorageMssql.test.ts @@ -0,0 +1,237 @@ +import { assert, describe, it } from "@effect/vitest" +import { Effect, Layer } from "effect" +import { MessageStorage, ShardingConfig, Snowflake, SqlMessageStorage } from "effect/unstable/cluster" +import { SqlClient } from "effect/unstable/sql" +import { MssqlContainer } from "../fixtures/mssql-utils.ts" + +const storageLive = (prefix: string) => + SqlMessageStorage.layerWith({ prefix }).pipe( + Layer.provideMerge(Snowflake.layerGenerator), + Layer.provide(ShardingConfig.layerDefaults) + ) + +describe("SqlMessageStorage mssql migrations", () => { + it.layer(Layer.orDie(MssqlContainer.layerClient), { + timeout: 120000 + })("mssql", (it) => { + it.effect("creates filtered unique indexes on a fresh database", () => + Effect.gen(function*() { + const sql = yield* SqlClient.SqlClient + + // building the storage layer runs the migrations + yield* Effect.gen(function*() { + yield* MessageStorage.MessageStorage + }).pipe(Effect.provide(storageLive("fresh"))) + + // rows the pre-fix constraints rejected: message_id NULL twice, chunk + // replies with kind NULL twice, NULL payload and headers parameters + // (the client binds NULL parameters as BIT) + yield* sql` + INSERT INTO fresh_messages (id, message_id, shard_id, entity_type, entity_id, kind, payload, headers, processed, request_id) + VALUES (1, NULL, 'shard[1]', 'test', '1', 0, ${null}, ${null}, 0, 1) + ` + yield* sql` + INSERT INTO fresh_messages (id, message_id, shard_id, entity_type, entity_id, kind, payload, headers, processed, request_id) + VALUES (2, NULL, 'shard[1]', 'test', '2', 0, ${null}, ${null}, 0, 2) + ` + yield* sql` + INSERT INTO fresh_replies (id, kind, request_id, payload, sequence, acked) + VALUES (1, NULL, 1, '{"chunk":0}', 0, 0) + ` + yield* sql` + INSERT INTO fresh_replies (id, kind, request_id, payload, sequence, acked) + VALUES (2, NULL, 1, '{"chunk":1}', 1, 0) + ` + + const uniqueConstraints = yield* sql<{ n: number }>` + SELECT COUNT(*) AS n FROM sys.key_constraints + WHERE type = 'UQ' + AND parent_object_id IN (OBJECT_ID(N'fresh_messages'), OBJECT_ID(N'fresh_replies')) + ` + assert.strictEqual(Number(uniqueConstraints[0].n), 0) + + const filteredIndexes = yield* sql<{ name: string }>` + SELECT name FROM sys.indexes + WHERE is_unique = 1 AND has_filter = 1 + AND ( + (object_id = OBJECT_ID(N'fresh_messages') AND name = 'fresh_messages_message_id_idx') + OR (object_id = OBJECT_ID(N'fresh_replies') AND name IN ('fresh_replies_one_exit', 'fresh_replies_sequence')) + ) + ` + assert.deepStrictEqual(filteredIndexes.map((row) => row.name).sort(), [ + "fresh_messages_message_id_idx", + "fresh_replies_one_exit", + "fresh_replies_sequence" + ]) + + const textColumns = yield* sql<{ n: number }>` + SELECT COUNT(*) AS n FROM sys.columns + WHERE object_id IN (OBJECT_ID(N'fresh_messages'), OBJECT_ID(N'fresh_replies')) + AND system_type_id = TYPE_ID(N'text') + ` + assert.strictEqual(Number(textColumns[0].n), 0) + }), 120000) + + it.effect("converges an existing pre-fix schema and is idempotent", () => + Effect.gen(function*() { + const sql = yield* SqlClient.SqlClient + + // stage the schema exactly as the pre-fix 0001 and 0002 migrations + // left it: inline unique constraints (one anonymous), TEXT columns + // and a migration history that ends at 0002 + yield* sql` + CREATE TABLE upgrade_messages ( + id BIGINT PRIMARY KEY, + rowid BIGINT IDENTITY(1,1), + message_id VARCHAR(255), + shard_id VARCHAR(50) NOT NULL, + entity_type VARCHAR(150) NOT NULL, + entity_id VARCHAR(255) NOT NULL, + kind INT NOT NULL, + tag VARCHAR(50), + payload TEXT, + headers TEXT, + trace_id VARCHAR(32), + span_id VARCHAR(16), + sampled BIT, + processed BIT NOT NULL DEFAULT 0, + request_id BIGINT NOT NULL, + reply_id BIGINT, + last_reply_id BIGINT, + last_read DATETIME, + deliver_at BIGINT, + UNIQUE (message_id) + ) + ` + yield* sql` + CREATE TABLE upgrade_replies ( + id BIGINT PRIMARY KEY, + rowid BIGINT IDENTITY(1,1), + kind INT, + request_id BIGINT NOT NULL, + payload TEXT NOT NULL, + sequence INT, + acked BIT NOT NULL DEFAULT 0, + CONSTRAINT upgrade_replies_one_exit UNIQUE (request_id, kind), + CONSTRAINT upgrade_replies_sequence UNIQUE (request_id, sequence) + ) + ` + yield* sql` + CREATE TABLE upgrade_migrations ( + migration_id INT NOT NULL PRIMARY KEY, + name VARCHAR(255) NOT NULL, + created_at DATETIME NOT NULL DEFAULT GETDATE() + ) + ` + yield* sql` + INSERT INTO upgrade_migrations (migration_id, name) + VALUES (1, 'create_tables'), (2, 'entity_type_size') + ` + // pre-existing data that must survive the TEXT to NVARCHAR(MAX) conversion + yield* sql` + INSERT INTO upgrade_messages (id, message_id, shard_id, entity_type, entity_id, kind, tag, payload, headers, processed, request_id) + VALUES (1, NULL, 'legacy[1]', 'test', '1', 0, 'GetUser', '{"id":1}', '{}', 1, 1) + ` + yield* sql` + INSERT INTO upgrade_replies (id, kind, request_id, payload, sequence, acked) + VALUES (1, 0, 1, '{"legacy":true}', NULL, 1) + ` + + // building the layer runs the remaining 0003 migration + yield* Effect.gen(function*() { + yield* MessageStorage.MessageStorage + }).pipe(Effect.provide(storageLive("upgrade"))) + + // rows the pre-fix constraints rejected: message_id NULL twice, chunk + // replies with kind NULL twice, NULL payload and headers parameters + // (the client binds NULL parameters as BIT, which TEXT columns reject) + yield* sql` + INSERT INTO upgrade_messages (id, message_id, shard_id, entity_type, entity_id, kind, payload, headers, processed, request_id) + VALUES (10, NULL, 'legacy[1]', 'test', '10', 0, ${null}, ${null}, 0, 10) + ` + yield* sql` + INSERT INTO upgrade_messages (id, message_id, shard_id, entity_type, entity_id, kind, payload, headers, processed, request_id) + VALUES (11, NULL, 'legacy[1]', 'test', '11', 0, ${null}, ${null}, 0, 11) + ` + yield* sql` + INSERT INTO upgrade_replies (id, kind, request_id, payload, sequence, acked) + VALUES (10, NULL, 10, '{"chunk":0}', 0, 0) + ` + yield* sql` + INSERT INTO upgrade_replies (id, kind, request_id, payload, sequence, acked) + VALUES (11, NULL, 10, '{"chunk":1}', 1, 0) + ` + + // the replacement indexes still enforce uniqueness of non-NULL keys + yield* sql` + INSERT INTO upgrade_messages (id, message_id, shard_id, entity_type, entity_id, kind, processed, request_id) + VALUES (90, 'dup-key', 'legacy[1]', 'test', '1', 0, 1, 90) + ` + const messageIdViolation = yield* Effect.flip(sql` + INSERT INTO upgrade_messages (id, message_id, shard_id, entity_type, entity_id, kind, processed, request_id) + VALUES (91, 'dup-key', 'legacy[1]', 'test', '1', 0, 1, 91) + `) + assert.strictEqual(messageIdViolation._tag, "SqlError") + const exitViolation = yield* Effect.flip(sql` + INSERT INTO upgrade_replies (id, kind, request_id, payload, sequence, acked) + VALUES (92, 0, 1, '{}', NULL, 1) + `) + assert.strictEqual(exitViolation._tag, "SqlError") + // duplicate non-NULL chunk sequence for the same request + yield* sql` + INSERT INTO upgrade_replies (id, kind, request_id, payload, sequence, acked) + VALUES (93, NULL, 90, '{"chunk":true}', 5, 0) + ` + const sequenceViolation = yield* Effect.flip(sql` + INSERT INTO upgrade_replies (id, kind, request_id, payload, sequence, acked) + VALUES (94, NULL, 90, '{"chunk":true}', 5, 0) + `) + assert.strictEqual(sequenceViolation._tag, "SqlError") + + const assertConverged = Effect.gen(function*() { + const uniqueConstraints = yield* sql<{ n: number }>` + SELECT COUNT(*) AS n FROM sys.key_constraints + WHERE type = 'UQ' + AND parent_object_id IN (OBJECT_ID(N'upgrade_messages'), OBJECT_ID(N'upgrade_replies')) + ` + assert.strictEqual(Number(uniqueConstraints[0].n), 0) + + const filteredIndexes = yield* sql<{ n: number }>` + SELECT COUNT(*) AS n FROM sys.indexes + WHERE is_unique = 1 AND has_filter = 1 + AND ( + (object_id = OBJECT_ID(N'upgrade_messages') AND name = 'upgrade_messages_message_id_idx') + OR (object_id = OBJECT_ID(N'upgrade_replies') AND name IN ('upgrade_replies_one_exit', 'upgrade_replies_sequence')) + ) + ` + assert.strictEqual(Number(filteredIndexes[0].n), 3) + + const textColumns = yield* sql<{ n: number }>` + SELECT COUNT(*) AS n FROM sys.columns + WHERE object_id IN (OBJECT_ID(N'upgrade_messages'), OBJECT_ID(N'upgrade_replies')) + AND system_type_id = TYPE_ID(N'text') + ` + assert.strictEqual(Number(textColumns[0].n), 0) + }) + yield* assertConverged + + // pre-existing data survived the column conversion + const messages = yield* sql<{ payload: string; headers: string }>` + SELECT payload, headers FROM upgrade_messages WHERE id = 1 + ` + assert.strictEqual(messages[0].payload, `{"id":1}`) + assert.strictEqual(messages[0].headers, "{}") + const replies = yield* sql<{ payload: string }>` + SELECT payload FROM upgrade_replies WHERE id = 1 + ` + assert.strictEqual(replies[0].payload, `{"legacy":true}`) + + // re-running the migration against the converged schema is harmless + yield* sql`DELETE FROM upgrade_migrations WHERE migration_id = 3` + yield* Effect.gen(function*() { + yield* MessageStorage.MessageStorage + }).pipe(Effect.provide(storageLive("upgrade"))) + yield* assertConverged + }), 120000) + }) +}) diff --git a/packages/platform-node/test/fixtures/mssql-utils.ts b/packages/platform-node/test/fixtures/mssql-utils.ts new file mode 100644 index 00000000000..ec82aa7a8d4 --- /dev/null +++ b/packages/platform-node/test/fixtures/mssql-utils.ts @@ -0,0 +1,41 @@ +import { MssqlClient } from "@effect/sql-mssql" +import type { StartedMSSQLServerContainer } from "@testcontainers/mssqlserver" +import { MSSQLServerContainer } from "@testcontainers/mssqlserver" +import { Context, Data, Effect, Layer, Redacted } from "effect" + +export class ContainerError extends Data.TaggedError("ContainerError")<{ + cause: unknown +}> {} + +export class MssqlContainer extends Context.Service< + MssqlContainer, + StartedMSSQLServerContainer +>()("test/MssqlContainer") { + static readonly layer = Layer.effect(this)( + Effect.acquireRelease( + Effect.tryPromise({ + try: () => + new MSSQLServerContainer("mcr.microsoft.com/mssql/server:2022-latest") + .acceptLicense() + .start(), + catch: (cause) => new ContainerError({ cause }) + }), + (container) => Effect.promise(() => container.stop()) + ) + ) + + static client = Layer.unwrap( + Effect.gen(function*() { + const container = yield* MssqlContainer + return MssqlClient.layer({ + server: container.getHost(), + port: container.getPort(), + username: container.getUsername(), + password: Redacted.make(container.getPassword()), + database: container.getDatabase() + }) + }) + ) + + static layerClient = this.client.pipe(Layer.provide(this.layer)) +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index a50b9a1b475..d6cdf3f0bde 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -481,6 +481,9 @@ importers: specifier: ^8.7.0 version: 8.7.0 devDependencies: + '@testcontainers/mssqlserver': + specifier: ^11.14.0 + version: 11.14.0 '@testcontainers/mysql': specifier: ^11.14.0 version: 11.14.0 @@ -2785,6 +2788,9 @@ packages: '@standard-schema/spec@1.1.0': resolution: {integrity: sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w==} + '@testcontainers/mssqlserver@11.14.0': + resolution: {integrity: sha512-/qjtcAg0UNIPmlh7ZQ+xsQE0Gpdw2gndg8shPhvCxxpCACW5Pw/dR6Hp87dcGZz551Ekr/r9fu3rUW852/wYhw==} + '@testcontainers/mysql@11.14.0': resolution: {integrity: sha512-0/OKd1gOvnl0qS+RIpmT6J0XKWpjkVyIbOtS3cFTVR8t6Ps7W9TV0U+zp7FPxCmitWuWYPXwH/+RQXF+1jZuZQ==} @@ -8413,6 +8419,15 @@ snapshots: '@standard-schema/spec@1.1.0': {} + '@testcontainers/mssqlserver@11.14.0': + dependencies: + testcontainers: 11.14.0 + transitivePeerDependencies: + - bare-abort-controller + - bare-buffer + - react-native-b4a + - supports-color + '@testcontainers/mysql@11.14.0': dependencies: testcontainers: 11.14.0