From 67a0952f12b8f91f2a92235af1f546546dbbcaae Mon Sep 17 00:00:00 2001 From: Ankit Varshney Date: Fri, 24 Jul 2026 02:53:27 +0530 Subject: [PATCH] Batch SQL persistence expiration cleanup --- .../batch-persistence-expiration-cleanup.md | 5 + packages/effect/src/internal/persistence.ts | 2 + .../src/unstable/persistence/Persistence.ts | 129 +++++++++++++++++- .../unstable/persistence/SqlCleanupTest.ts | 20 +++ packages/sql/mysql2/test/Persistence.test.ts | 65 ++++++++- packages/sql/pg/test/Persistence.test.ts | 59 ++++++++ .../sql/sqlite-node/test/Persistence.test.ts | 63 ++++++++- 7 files changed, 335 insertions(+), 8 deletions(-) create mode 100644 .changeset/batch-persistence-expiration-cleanup.md create mode 100644 packages/effect/src/internal/persistence.ts create mode 100644 packages/effect/test/unstable/persistence/SqlCleanupTest.ts diff --git a/.changeset/batch-persistence-expiration-cleanup.md b/.changeset/batch-persistence-expiration-cleanup.md new file mode 100644 index 00000000000..3409edfb689 --- /dev/null +++ b/.changeset/batch-persistence-expiration-cleanup.md @@ -0,0 +1,5 @@ +--- +"effect": patch +--- + +Run shared-table SQL persistence expiration cleanup in indexed, bounded background batches. diff --git a/packages/effect/src/internal/persistence.ts b/packages/effect/src/internal/persistence.ts new file mode 100644 index 00000000000..5d222c257ab --- /dev/null +++ b/packages/effect/src/internal/persistence.ts @@ -0,0 +1,2 @@ +/** @internal */ +export const sqlCleanupBatchSize = 1000 diff --git a/packages/effect/src/unstable/persistence/Persistence.ts b/packages/effect/src/unstable/persistence/Persistence.ts index 975aea8333e..a0d445a658d 100644 --- a/packages/effect/src/unstable/persistence/Persistence.ts +++ b/packages/effect/src/unstable/persistence/Persistence.ts @@ -15,8 +15,11 @@ import * as Duration from "../../Duration.ts" import * as Effect from "../../Effect.ts" import * as Exit from "../../Exit.ts" import { identity } from "../../Function.ts" +import { sqlCleanupBatchSize } from "../../internal/persistence.ts" import * as Layer from "../../Layer.ts" import * as PrimaryKey from "../../PrimaryKey.ts" +import * as Result from "../../Result.ts" +import * as Schedule from "../../Schedule.ts" import * as Schema from "../../Schema.ts" import type * as Scope from "../../Scope.ts" import * as SqlClient from "../sql/SqlClient.ts" @@ -560,6 +563,127 @@ export const layerBackingSql: Layer.Layer< ` }).pipe(Effect.orDie) + yield* sql.onDialectOrElse({ + pg: () => + sql`CREATE INDEX IF NOT EXISTS effect_persistence_expires_idx ON ${table} (expires) WHERE expires IS NOT NULL` + .pipe(Effect.orDie, Effect.asVoid), + mysql: () => + Effect.gen(function*() { + const indexExists = sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM information_schema.statistics + WHERE table_schema = DATABASE() + AND table_name = 'effect_persistence' + AND index_name = 'effect_persistence_expires_idx' + `.pipe( + Effect.map((rows) => Number(rows[0].count) > 0), + Effect.orDie + ) + if (yield* indexExists) return + + const createIndexResult = yield* sql`CREATE INDEX effect_persistence_expires_idx ON ${table} (expires)` + .pipe(Effect.result) + if (!(yield* indexExists)) { + if (Result.isFailure(createIndexResult)) { + return yield* Effect.die(createIndexResult.failure) + } + return yield* Effect.die(new Error("Failed to create effect_persistence_expires_idx")) + } + }), + mssql: () => + sql` + IF NOT EXISTS ( + SELECT * FROM sys.indexes + WHERE name = N'effect_persistence_expires_idx' + AND object_id = OBJECT_ID(N'effect_persistence') + ) + CREATE INDEX effect_persistence_expires_idx ON ${table} (expires) WHERE expires IS NOT NULL + `.pipe(Effect.orDie, Effect.asVoid), + // sqlite + orElse: () => + sql`CREATE INDEX IF NOT EXISTS effect_persistence_expires_idx ON ${table} (expires) WHERE expires IS NOT NULL` + .pipe(Effect.orDie, Effect.asVoid) + }) + + const cleanupBatchDelay = Duration.millis(10) + const cleanupInterval = Duration.minutes(5) + + const deleteExpiredBatch = sql.onDialectOrElse({ + pg: () => (expiresAtOrBefore: number) => + sql<{ readonly tupleId: string }>` + WITH expired_entries AS ( + SELECT ctid FROM ${table} + WHERE expires IS NOT NULL AND expires <= ${expiresAtOrBefore} + ORDER BY expires + LIMIT ${sql.literal(String(sqlCleanupBatchSize))} + ) + DELETE FROM ${table} + WHERE ctid IN (SELECT ctid FROM expired_entries) + RETURNING ctid::text AS "tupleId" + `.pipe(Effect.map((deletedEntries) => deletedEntries.length)), + mysql: () => + Effect.fnUntraced( + function*(expiresAtOrBefore: number) { + yield* sql` + DELETE FROM ${table} + WHERE expires IS NOT NULL AND expires <= ${expiresAtOrBefore} + ORDER BY expires + LIMIT ${sql.literal(String(sqlCleanupBatchSize))} + ` + const rows = yield* sql<{ readonly count: number }>`SELECT ROW_COUNT() AS count` + return Number(rows[0].count) + }, + (effect) => effect.pipe(sql.withTransaction) + ), + mssql: () => (expiresAtOrBefore: number) => + sql<{ readonly store_id: string }>` + WITH expired_entries AS ( + SELECT TOP ${sql.literal(String(sqlCleanupBatchSize))} store_id, id FROM ${table} + WITH (UPDLOCK, READPAST, READCOMMITTEDLOCK) + WHERE expires IS NOT NULL AND expires <= ${expiresAtOrBefore} + ORDER BY expires + ) + DELETE persistence + OUTPUT DELETED.store_id + FROM ${table} AS persistence + INNER JOIN expired_entries + ON persistence.store_id = expired_entries.store_id + AND persistence.id = expired_entries.id + `.pipe(Effect.map((deletedEntries) => deletedEntries.length)), + // Some sqlite clients do not support interactive transactions, so use one bounded statement. + orElse: () => (expiresAtOrBefore: number) => + sql<{ readonly store_id: string }>` + DELETE FROM ${table} + WHERE rowid IN ( + SELECT rowid FROM ${table} + WHERE expires IS NOT NULL AND expires <= ${expiresAtOrBefore} + ORDER BY expires + LIMIT ${sql.literal(String(sqlCleanupBatchSize))} + ) + RETURNING store_id + `.pipe(Effect.map((deletedEntries) => deletedEntries.length)) + }) + + // Delay between batches and stop once a batch deletes no rows. + const deleteExpiredSchedule = Schedule.forever.pipe( + Schedule.setInputType(), + Schedule.passthrough, + Schedule.while(({ input: deletedCount }) => deletedCount !== 0), + Schedule.addDelay(() => Effect.succeed(cleanupBatchDelay)) + ) + + const deleteExpired = Effect.fnUntraced(function*() { + const expiresAtOrBefore = yield* Clock.currentTimeMillis + return yield* deleteExpiredBatch(expiresAtOrBefore).pipe( + Effect.repeat(deleteExpiredSchedule) + ) + }) + + yield* deleteExpired().pipe( + Effect.catch((cause) => Effect.logWarning("Failed to clean up expired persistence entries", cause)), + Effect.repeat(Schedule.spaced(cleanupInterval)), + Effect.forkScoped + ) + type UpsertFn = ( entries: Array<{ store_id: string; id: string; value: string; expires: number | null }> ) => Effect.Effect @@ -606,11 +730,6 @@ export const layerBackingSql: Layer.Layer< make: Effect.fnUntraced(function*(storeId) { const clock = yield* Clock.Clock - // Cleanup expired entries on startup - yield* Effect.ignore( - sql`DELETE FROM ${table} WHERE store_id = ${storeId} AND expires IS NOT NULL AND expires <= ${clock.currentTimeMillisUnsafe()}` - ) - return identity({ get: (key) => sql< diff --git a/packages/effect/test/unstable/persistence/SqlCleanupTest.ts b/packages/effect/test/unstable/persistence/SqlCleanupTest.ts new file mode 100644 index 00000000000..77accf3f88f --- /dev/null +++ b/packages/effect/test/unstable/persistence/SqlCleanupTest.ts @@ -0,0 +1,20 @@ +import { Duration, Effect, Schedule } from "effect" +import { sqlCleanupBatchSize } from "effect/internal/persistence" +import { TestClock } from "effect/testing" + +export const expiredEntryCount = sqlCleanupBatchSize + 1 +export const cleanupBatchDelay = Duration.millis(10) +export const testTimeout = 30_000 + +export const waitForCount = ( + effect: Effect.Effect, + predicate: (count: number) => boolean +) => + effect.pipe( + Effect.repeat({ + until: predicate, + schedule: Schedule.spaced(cleanupBatchDelay) + }), + Effect.timeout(Duration.millis(testTimeout)), + TestClock.withLive + ) diff --git a/packages/sql/mysql2/test/Persistence.test.ts b/packages/sql/mysql2/test/Persistence.test.ts index b0e9b76250a..9623cbc55c1 100644 --- a/packages/sql/mysql2/test/Persistence.test.ts +++ b/packages/sql/mysql2/test/Persistence.test.ts @@ -1,7 +1,11 @@ -import { Layer } from "effect" +import { assert, it } from "@effect/vitest" +import { Effect, Layer } from "effect" import * as PersistedCacheTest from "effect-test/unstable/persistence/PersistedCacheTest" import * as PersistedQueueTest from "effect-test/unstable/persistence/PersistedQueueTest" +import * as SqlCleanupTest from "effect-test/unstable/persistence/SqlCleanupTest" +import { TestClock } from "effect/testing" import { PersistedQueue, Persistence } from "effect/unstable/persistence" +import { SqlClient } from "effect/unstable/sql" import { MysqlContainer } from "./utils.ts" PersistedCacheTest.suite( @@ -20,3 +24,62 @@ PersistedQueueTest.suite( Layer.provide(MysqlContainer.layerClient) ) ) + +it.layer(MysqlContainer.layerClient, { timeout: "30 seconds" })("Persistence SQL cleanup", (it) => { + it.effect("deletes expired entries in batches", () => + Effect.gen(function*() { + const sql = (yield* SqlClient.SqlClient).withoutTransforms() + const table = sql("effect_persistence") + const expiredCount = sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM ${table} WHERE store_id = 'expired' + `.pipe(Effect.map((rows) => Number(rows[0].count))) + yield* sql` + CREATE TABLE ${table} ( + store_id VARCHAR(191) NOT NULL, + id VARCHAR(191) NOT NULL, + value TEXT NOT NULL, + expires BIGINT, + PRIMARY KEY (store_id, id) + ) + ` + + const entries = Array.from({ length: SqlCleanupTest.expiredEntryCount }, (_, i) => ({ + store_id: "expired", + id: String(i), + value: "{}", + expires: 0 + })) + yield* sql`INSERT INTO ${table} ${sql.insert(entries)}`.unprepared + yield* sql` + INSERT INTO ${table} (store_id, id, value, expires) + VALUES ('live', 'live', '{}', NULL), ('live', 'future', '{}', 1000000) + ` + + yield* Layer.build(Persistence.layerBackingSql) + + let expired = yield* SqlCleanupTest.waitForCount( + expiredCount, + (count) => count < SqlCleanupTest.expiredEntryCount + ) + assert.strictEqual(expired, 1) + + while (expired > 0) { + const previous = expired + yield* TestClock.adjust(SqlCleanupTest.cleanupBatchDelay) + expired = yield* SqlCleanupTest.waitForCount(expiredCount, (count) => count < previous) + } + assert.strictEqual(expired, 0) + const live = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM ${table} WHERE store_id = 'live' + ` + assert.strictEqual(Number(live[0].count), 2) + + const indexes = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM information_schema.statistics + WHERE table_schema = DATABASE() + AND table_name = 'effect_persistence' + AND index_name = 'effect_persistence_expires_idx' + ` + assert.strictEqual(Number(indexes[0].count), 1) + }), { timeout: SqlCleanupTest.testTimeout }) +}) diff --git a/packages/sql/pg/test/Persistence.test.ts b/packages/sql/pg/test/Persistence.test.ts index 4ee287631cc..ee77d72748a 100644 --- a/packages/sql/pg/test/Persistence.test.ts +++ b/packages/sql/pg/test/Persistence.test.ts @@ -2,6 +2,7 @@ import { assert, it } from "@effect/vitest" import { Effect, Exit, Fiber, Latch, Layer, Schema } from "effect" import * as PersistedCacheTest from "effect-test/unstable/persistence/PersistedCacheTest" import * as PersistedQueueTest from "effect-test/unstable/persistence/PersistedQueueTest" +import * as SqlCleanupTest from "effect-test/unstable/persistence/SqlCleanupTest" import { TestClock } from "effect/testing" import { PersistedQueue, Persistence } from "effect/unstable/persistence" import { SqlClient } from "effect/unstable/sql" @@ -104,3 +105,61 @@ it.layer(PgContainer.layerClient, { timeout: "30 seconds" })("PersistedQueue SQL assert.strictEqual(value, "valid") }).pipe(TestClock.withLive)) }) + +it.layer(PgContainer.layerClient, { timeout: "30 seconds" })("Persistence SQL cleanup", (it) => { + it.effect("deletes expired entries in batches", () => + Effect.gen(function*() { + const sql = (yield* SqlClient.SqlClient).withoutTransforms() + const table = sql("effect_persistence") + const expiredCount = sql<{ readonly count: number }>` + SELECT COUNT(*)::INT AS count FROM ${table} WHERE store_id = 'expired' + `.pipe(Effect.map((rows) => rows[0].count)) + yield* sql` + CREATE TABLE ${table} ( + store_id TEXT NOT NULL, + id TEXT NOT NULL, + value TEXT NOT NULL, + expires BIGINT, + PRIMARY KEY (store_id, id) + ) + ` + + const entries = Array.from({ length: SqlCleanupTest.expiredEntryCount }, (_, i) => ({ + store_id: "expired", + id: String(i), + value: "{}", + expires: 0 + })) + yield* sql`INSERT INTO ${table} ${sql.insert(entries)}`.unprepared + yield* sql` + INSERT INTO ${table} (store_id, id, value, expires) + VALUES ('live', 'live', '{}', NULL), ('live', 'future', '{}', 1000000) + ` + + yield* Layer.build(Persistence.layerBackingSql) + + let expired = yield* SqlCleanupTest.waitForCount( + expiredCount, + (count) => count < SqlCleanupTest.expiredEntryCount + ) + assert.strictEqual(expired, 1) + + while (expired > 0) { + const previous = expired + yield* TestClock.adjust(SqlCleanupTest.cleanupBatchDelay) + expired = yield* SqlCleanupTest.waitForCount(expiredCount, (count) => count < previous) + } + assert.strictEqual(expired, 0) + const live = yield* sql<{ readonly count: number }>` + SELECT COUNT(*)::INT AS count FROM ${table} WHERE store_id = 'live' + ` + assert.strictEqual(live[0].count, 2) + + const indexes = yield* sql<{ readonly count: number }>` + SELECT COUNT(*)::INT AS count FROM pg_indexes + WHERE tablename = 'effect_persistence' + AND indexname = 'effect_persistence_expires_idx' + ` + assert.strictEqual(indexes[0].count, 1) + }), { timeout: SqlCleanupTest.testTimeout }) +}) diff --git a/packages/sql/sqlite-node/test/Persistence.test.ts b/packages/sql/sqlite-node/test/Persistence.test.ts index 78313b98a02..14f8ca7172d 100644 --- a/packages/sql/sqlite-node/test/Persistence.test.ts +++ b/packages/sql/sqlite-node/test/Persistence.test.ts @@ -1,11 +1,12 @@ import { NodeFileSystem } from "@effect/platform-node" import { SqliteClient } from "@effect/sql-sqlite-node" -import { expect, it } from "@effect/vitest" +import { assert, expect, it } from "@effect/vitest" import { Duration, Effect, FileSystem, Layer } from "effect" +import * as SqlCleanupTest from "effect-test/unstable/persistence/SqlCleanupTest" import { TestClock } from "effect/testing" import { Persistence } from "effect/unstable/persistence" import { Reactivity } from "effect/unstable/reactivity" -import type * as SqlClient from "effect/unstable/sql/SqlClient" +import * as SqlClient from "effect/unstable/sql/SqlClient" const ClientLayer = Effect.gen(function*() { const fs = yield* FileSystem.FileSystem @@ -107,3 +108,61 @@ const suite = (name: string, layer: Layer.Layer { + it.effect("deletes expired entries in batches", () => + Effect.gen(function*() { + const sql = (yield* SqlClient.SqlClient).withoutTransforms() + const table = sql("effect_persistence") + const expiredCount = sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM ${table} WHERE store_id = 'expired' + `.pipe(Effect.map((rows) => rows[0].count)) + yield* sql` + CREATE TABLE ${table} ( + store_id TEXT NOT NULL, + id TEXT NOT NULL, + value TEXT NOT NULL, + expires INTEGER, + PRIMARY KEY (store_id, id) + ) + ` + + const entries = Array.from({ length: SqlCleanupTest.expiredEntryCount }, (_, i) => ({ + store_id: "expired", + id: String(i), + value: "{}", + expires: 0 + })) + yield* sql`INSERT INTO ${table} ${sql.insert(entries)}`.unprepared + yield* sql` + INSERT INTO ${table} (store_id, id, value, expires) + VALUES ('live', 'live', '{}', NULL), ('live', 'future', '{}', 1000000) + ` + + yield* Layer.build(Persistence.layerBackingSql) + + let expired = yield* SqlCleanupTest.waitForCount( + expiredCount, + (count) => count < SqlCleanupTest.expiredEntryCount + ) + assert.strictEqual(expired, 1) + + while (expired > 0) { + const previous = expired + yield* TestClock.adjust(SqlCleanupTest.cleanupBatchDelay) + expired = yield* SqlCleanupTest.waitForCount(expiredCount, (count) => count < previous) + } + assert.strictEqual(expired, 0) + const live = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM ${table} WHERE store_id = 'live' + ` + assert.strictEqual(live[0].count, 2) + + const indexes = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM sqlite_master + WHERE type = 'index' + AND name = 'effect_persistence_expires_idx' + ` + assert.strictEqual(indexes[0].count, 1) + }), { timeout: SqlCleanupTest.testTimeout }) +})