From a29dd597cde8977d62103b062c3c423d00717db6 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 13 Jul 2026 13:34:13 +0100 Subject: [PATCH 1/3] feat(webapp): add option to disable PostgreSQL task-event writes Adds EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED (default off). When enabled, the task-event store skips all PostgreSQL writes, for deployments that store task events in ClickHouse (EVENT_REPOSITORY_DEFAULT_STORE=clickhouse_v2). Reads are unaffected, so existing events stay readable. --- .server-changes/disable-postgres-task-event-writes.md | 6 ++++++ apps/webapp/app/env.server.ts | 1 + .../app/v3/eventRepository/eventRepository.server.ts | 6 +++++- apps/webapp/app/v3/taskEventStore.server.ts | 9 ++++++++- docs/self-hosting/env/webapp.mdx | 1 + 5 files changed, 21 insertions(+), 2 deletions(-) create mode 100644 .server-changes/disable-postgres-task-event-writes.md diff --git a/.server-changes/disable-postgres-task-event-writes.md b/.server-changes/disable-postgres-task-event-writes.md new file mode 100644 index 00000000000..47eb9ef9fd1 --- /dev/null +++ b/.server-changes/disable-postgres-task-event-writes.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: feature +--- + +Added `EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED` to skip all PostgreSQL task-event writes for deployments that store task events in ClickHouse. Leave it off unless `EVENT_REPOSITORY_DEFAULT_STORE` is `clickhouse_v2`, otherwise task events are lost. diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 6dc6ccee3f2..0e2b5ab78fe 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -1828,6 +1828,7 @@ const EnvironmentSchema = z .enum(["postgres", "clickhouse", "clickhouse_v2"]) .default("postgres"), EVENT_REPOSITORY_DEBUG_LOGS_DISABLED: BoolEnv.default(false), + EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED: BoolEnv.default(false), EVENTS_CLICKHOUSE_MAX_TRACE_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(25_000), EVENTS_CLICKHOUSE_MAX_TRACE_DETAILED_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(5_000), EVENTS_CLICKHOUSE_MAX_LIVE_RELOADING_SETTING: z.coerce.number().int().default(2000), diff --git a/apps/webapp/app/v3/eventRepository/eventRepository.server.ts b/apps/webapp/app/v3/eventRepository/eventRepository.server.ts index 2de5ce8b25a..40c84bb1af2 100644 --- a/apps/webapp/app/v3/eventRepository/eventRepository.server.ts +++ b/apps/webapp/app/v3/eventRepository/eventRepository.server.ts @@ -120,7 +120,11 @@ export class EventRepository implements IEventRepository { this._tracer = _config.tracer ?? trace.getTracer("eventRepo", "0.0.1"); // Instantiate the store using the partitioning flag. - this.taskEventStore = new TaskEventStore(db, readReplica); + this.taskEventStore = new TaskEventStore( + db, + readReplica, + env.EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED + ); } #createableEventToPrismaEvent(event: CreateEventInput): Prisma.TaskEventCreateManyInput { diff --git a/apps/webapp/app/v3/taskEventStore.server.ts b/apps/webapp/app/v3/taskEventStore.server.ts index a92db8d4284..83403861971 100644 --- a/apps/webapp/app/v3/taskEventStore.server.ts +++ b/apps/webapp/app/v3/taskEventStore.server.ts @@ -59,13 +59,17 @@ export function getTaskEventStore(): TaskEventStoreTable { export class TaskEventStore { constructor( private db: PrismaClient, - private readReplica: PrismaReplicaClient + private readReplica: PrismaReplicaClient, + private writesDisabled: boolean = false ) {} /** * Insert one record. */ async create(table: TaskEventStoreTable, data: Prisma.TaskEventCreateInput) { + if (this.writesDisabled) { + return; + } if (table === "taskEventPartitioned") { return await this.db.taskEventPartitioned.create({ data }); } else { @@ -77,6 +81,9 @@ export class TaskEventStore { * Insert many records. */ async createMany(table: TaskEventStoreTable, data: Prisma.TaskEventCreateManyInput[]) { + if (this.writesDisabled) { + return { count: 0 }; + } if (table === "taskEventPartitioned") { return await this.db.taskEventPartitioned.createMany({ data }); } else { diff --git a/docs/self-hosting/env/webapp.mdx b/docs/self-hosting/env/webapp.mdx index d4e8b16ff39..1f629718b52 100644 --- a/docs/self-hosting/env/webapp.mdx +++ b/docs/self-hosting/env/webapp.mdx @@ -135,6 +135,7 @@ mode: "wide" | `SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT` | No | 8192 | OTel span attribute value length limit. | | **Task events** | | | | | `EVENT_REPOSITORY_DEFAULT_STORE` | No | postgres | Where to store task events. Set to `clickhouse_v2` to store in ClickHouse (recommended for production). | +| `EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED` | No | 0 | Skip all PostgreSQL task-event writes (set to `1`). Only enable when `EVENT_REPOSITORY_DEFAULT_STORE` is `clickhouse_v2`, otherwise task events are lost. | | **Realtime** | | | | | `REALTIME_STREAM_VERSION` | No | v1 | Stream version exposed to tasks via the `TRIGGER_REALTIME_STREAM_VERSION` variable. Distinct from `REALTIME_STREAMS_DEFAULT_VERSION`. One of `v1`, `v2`. | | `REALTIME_STREAM_MAX_LENGTH` | No | 1000 | Realtime stream max length. | From c46563a333fcfd5c4f02b8da081ff4feb896e54d Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 13 Jul 2026 14:01:53 +0100 Subject: [PATCH 2/3] refactor(webapp): move the postgres task-event write guard to the repository entry points Guarding at TaskEventStore.createMany stopped the write but the repository still buffered, reported success, and published Redis notifications for events that were never persisted. Move the check to the insertMany/insertImmediate/ insertManyImmediate entry points so the postgres store is fully inert when writes are disabled. --- .../v3/eventRepository/eventRepository.server.ts | 15 ++++++++++----- apps/webapp/app/v3/taskEventStore.server.ts | 9 +-------- 2 files changed, 11 insertions(+), 13 deletions(-) diff --git a/apps/webapp/app/v3/eventRepository/eventRepository.server.ts b/apps/webapp/app/v3/eventRepository/eventRepository.server.ts index 40c84bb1af2..588a49737af 100644 --- a/apps/webapp/app/v3/eventRepository/eventRepository.server.ts +++ b/apps/webapp/app/v3/eventRepository/eventRepository.server.ts @@ -120,11 +120,7 @@ export class EventRepository implements IEventRepository { this._tracer = _config.tracer ?? trace.getTracer("eventRepo", "0.0.1"); // Instantiate the store using the partitioning flag. - this.taskEventStore = new TaskEventStore( - db, - readReplica, - env.EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED - ); + this.taskEventStore = new TaskEventStore(db, readReplica); } #createableEventToPrismaEvent(event: CreateEventInput): Prisma.TaskEventCreateManyInput { @@ -161,14 +157,23 @@ export class EventRepository implements IEventRepository { } private async insertImmediate(event: CreateEventInput) { + if (env.EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED) { + return; + } await this.#flushBatch(nanoid(), [this.#createableEventToPrismaEvent(event)]); } insertMany(events: CreateEventInput[]) { + if (env.EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED) { + return; + } this._flushScheduler.addToBatch(events.map(this.#createableEventToPrismaEvent)); } async insertManyImmediate(events: CreateEventInput[]) { + if (env.EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED) { + return; + } await this.#flushBatchWithReturn(nanoid(), events.map(this.#createableEventToPrismaEvent)); } diff --git a/apps/webapp/app/v3/taskEventStore.server.ts b/apps/webapp/app/v3/taskEventStore.server.ts index 83403861971..a92db8d4284 100644 --- a/apps/webapp/app/v3/taskEventStore.server.ts +++ b/apps/webapp/app/v3/taskEventStore.server.ts @@ -59,17 +59,13 @@ export function getTaskEventStore(): TaskEventStoreTable { export class TaskEventStore { constructor( private db: PrismaClient, - private readReplica: PrismaReplicaClient, - private writesDisabled: boolean = false + private readReplica: PrismaReplicaClient ) {} /** * Insert one record. */ async create(table: TaskEventStoreTable, data: Prisma.TaskEventCreateInput) { - if (this.writesDisabled) { - return; - } if (table === "taskEventPartitioned") { return await this.db.taskEventPartitioned.create({ data }); } else { @@ -81,9 +77,6 @@ export class TaskEventStore { * Insert many records. */ async createMany(table: TaskEventStoreTable, data: Prisma.TaskEventCreateManyInput[]) { - if (this.writesDisabled) { - return { count: 0 }; - } if (table === "taskEventPartitioned") { return await this.db.taskEventPartitioned.createMany({ data }); } else { From 5168714c86ced1a820baa322632b00e8b0bd3e28 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 13 Jul 2026 14:11:28 +0100 Subject: [PATCH 3/3] fix(webapp): route recordEvent/traceEvent batched writes through insertMany The postgres-writes-disabled guard was bypassed: recordEvent and traceEvent fed the flush scheduler directly for non-immediate events, so they still wrote to PostgreSQL with the flag on. Route both through the guarded insertMany, making it the only feeder of the flush scheduler. --- apps/webapp/app/v3/eventRepository/eventRepository.server.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/webapp/app/v3/eventRepository/eventRepository.server.ts b/apps/webapp/app/v3/eventRepository/eventRepository.server.ts index 588a49737af..ea778fb5356 100644 --- a/apps/webapp/app/v3/eventRepository/eventRepository.server.ts +++ b/apps/webapp/app/v3/eventRepository/eventRepository.server.ts @@ -1027,7 +1027,7 @@ export class EventRepository implements IEventRepository { if (options.immediate) { await this.insertImmediate(event); } else { - this._flushScheduler.addToBatch([this.#createableEventToPrismaEvent(event)]); + this.insertMany([event]); } } @@ -1161,7 +1161,7 @@ export class EventRepository implements IEventRepository { if (options.immediate) { await this.insertImmediate(event); } else { - this._flushScheduler.addToBatch([this.#createableEventToPrismaEvent(event)]); + this.insertMany([event]); } return result;