From 05b412cd280c94bc3c9145ec139ae09711a449c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ra=C3=BAl=20G=C3=B3mez=20Morales?= Date: Tue, 14 Jul 2026 13:12:31 +0200 Subject: [PATCH] feat(core): Add the Instance AI durable event-log store and writer (no-changelog) (#33910) Co-authored-by: Danny Martini Co-authored-by: Claude Fable 5 --- docs/generated/postgres-schema/README.md | 11 + .../public.instance_ai_events.md | 65 ++ .../public.instance_ai_threads.md | 12 +- docs/generated/sqlite-schema/README.md | 11 + .../sqlite-schema/instance_ai_events.md | 71 ++ .../sqlite-schema/instance_ai_threads.md | 12 +- ...84000000046-CreateInstanceAiEventsTable.ts | 36 + .../db/src/migrations/postgresdb/index.ts | 2 + .../@n8n/db/src/migrations/sqlite/index.ts | 2 + .../src/events/maps/instance-ai.event-map.ts | 26 + .../prometheus/instance-ai-metrics.service.ts | 71 ++ .../src/modules/instance-ai/entities/index.ts | 1 + .../instance-ai-event-log-entry.entity.ts | 42 ++ .../__tests__/durable-event-log.test.ts | 465 +++++++++++++ .../event-bus/durable-event-log.ts | 618 ++++++++++++++++++ .../event-bus/durable-log-metrics.ts | 90 +++ .../modules/instance-ai/instance-ai.module.ts | 4 + .../modules/instance-ai/repositories/index.ts | 1 + .../instance-ai-event-log.repository.ts | 80 +++ 19 files changed, 1618 insertions(+), 2 deletions(-) create mode 100644 docs/generated/postgres-schema/public.instance_ai_events.md create mode 100644 docs/generated/sqlite-schema/instance_ai_events.md create mode 100644 packages/@n8n/db/src/migrations/common/1784000000046-CreateInstanceAiEventsTable.ts create mode 100644 packages/cli/src/modules/instance-ai/entities/instance-ai-event-log-entry.entity.ts create mode 100644 packages/cli/src/modules/instance-ai/event-bus/__tests__/durable-event-log.test.ts create mode 100644 packages/cli/src/modules/instance-ai/event-bus/durable-event-log.ts create mode 100644 packages/cli/src/modules/instance-ai/event-bus/durable-log-metrics.ts create mode 100644 packages/cli/src/modules/instance-ai/repositories/instance-ai-event-log.repository.ts diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md index 87397680967..3eb9713f843 100644 --- a/docs/generated/postgres-schema/README.md +++ b/docs/generated/postgres-schema/README.md @@ -63,6 +63,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand. | [public.installed_nodes](public.installed_nodes.md) | 4 | | BASE TABLE | | [public.installed_packages](public.installed_packages.md) | 6 | | BASE TABLE | | [public.instance_ai_checkpoints](public.instance_ai_checkpoints.md) | 8 | | BASE TABLE | +| [public.instance_ai_events](public.instance_ai_events.md) | 7 | | BASE TABLE | | [public.instance_ai_iteration_logs](public.instance_ai_iteration_logs.md) | 6 | | BASE TABLE | | [public.instance_ai_mcp_registry_connections](public.instance_ai_mcp_registry_connections.md) | 7 | | BASE TABLE | | [public.instance_ai_messages](public.instance_ai_messages.md) | 8 | | BASE TABLE | @@ -227,6 +228,7 @@ erDiagram "public.insights_raw" }o--|| "public.insights_metadata" : "FOREIGN KEY (#quot;metaId#quot;) REFERENCES insights_metadata(#quot;metaId#quot;) ON DELETE CASCADE" "public.installed_nodes" }o--|| "public.installed_packages" : "FOREIGN KEY (package) REFERENCES installed_packages(#quot;packageName#quot;) ON UPDATE CASCADE ON DELETE CASCADE" "public.instance_ai_checkpoints" }o--|| "public.instance_ai_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES instance_ai_threads(id) ON DELETE CASCADE" +"public.instance_ai_events" }o--|| "public.instance_ai_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES instance_ai_threads(id) ON DELETE CASCADE" "public.instance_ai_iteration_logs" }o--|| "public.instance_ai_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES instance_ai_threads(id) ON DELETE CASCADE" "public.instance_ai_mcp_registry_connections" }o--|| "public.user" : "FOREIGN KEY (#quot;userId#quot;) REFERENCES #quot;user#quot;(id) ON DELETE CASCADE" "public.instance_ai_mcp_registry_connections" }o--|| "public.credentials_entity" : "FOREIGN KEY (#quot;credentialId#quot;) REFERENCES credentials_entity(id) ON DELETE CASCADE" @@ -824,6 +826,15 @@ erDiagram uuid threadId FK timestamp_3__with_time_zone updatedAt } +"public.instance_ai_events" { + timestamp_3__with_time_zone createdAt + text payload + varchar_64_ runId + integer seq + uuid threadId FK + varchar_64_ type + timestamp_3__with_time_zone updatedAt +} "public.instance_ai_iteration_logs" { timestamp_3__with_time_zone createdAt text entry diff --git a/docs/generated/postgres-schema/public.instance_ai_events.md b/docs/generated/postgres-schema/public.instance_ai_events.md new file mode 100644 index 00000000000..ca7341a8b80 --- /dev/null +++ b/docs/generated/postgres-schema/public.instance_ai_events.md @@ -0,0 +1,65 @@ +# public.instance_ai_events + +## Columns + +| Name | Type | Default | Nullable | Children | Parents | Comment | +| ---- | ---- | ------- | -------- | -------- | ------- | ------- | +| createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | +| payload | text | | false | | | JSON of the canonical InstanceAiEvent | +| runId | varchar(64) | | false | | | Run that emitted the event — opaque ID from the agent runtime | +| seq | integer | | false | | | Per-thread monotonic sequence — the SSE replay cursor | +| threadId | uuid | | false | | [public.instance_ai_threads](public.instance_ai_threads.md) | | +| type | varchar(64) | | false | | | Event type discriminator, duplicated out of the payload | +| updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | + +## Constraints + +| Name | Type | Definition | +| ---- | ---- | ---------- | +| FK_35909c5576a4a6c1d6a6fb71caa | FOREIGN KEY | FOREIGN KEY ("threadId") REFERENCES instance_ai_threads(id) ON DELETE CASCADE | +| PK_12489cd6197feeac2089acc7ef6 | PRIMARY KEY | PRIMARY KEY ("threadId", seq) | +| instance_ai_events_createdAt_not_null | n | NOT NULL "createdAt" | +| instance_ai_events_payload_not_null | n | NOT NULL payload | +| instance_ai_events_runId_not_null | n | NOT NULL "runId" | +| instance_ai_events_seq_not_null | n | NOT NULL seq | +| instance_ai_events_threadId_not_null | n | NOT NULL "threadId" | +| instance_ai_events_type_not_null | n | NOT NULL type | +| instance_ai_events_updatedAt_not_null | n | NOT NULL "updatedAt" | + +## Indexes + +| Name | Definition | +| ---- | ---------- | +| IDX_32cdd799675715fb1d2a8683e9 | CREATE INDEX "IDX_32cdd799675715fb1d2a8683e9" ON public.instance_ai_events USING btree ("threadId", "runId") | +| PK_12489cd6197feeac2089acc7ef6 | CREATE UNIQUE INDEX "PK_12489cd6197feeac2089acc7ef6" ON public.instance_ai_events USING btree ("threadId", seq) | + +## Relations + +```mermaid +erDiagram + +"public.instance_ai_events" }o--|| "public.instance_ai_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES instance_ai_threads(id) ON DELETE CASCADE" + +"public.instance_ai_events" { + timestamp_3__with_time_zone createdAt + text payload + varchar_64_ runId + integer seq + uuid threadId FK + varchar_64_ type + timestamp_3__with_time_zone updatedAt +} +"public.instance_ai_threads" { + timestamp_3__with_time_zone createdAt + uuid id + json metadata + varchar_36_ projectId FK + varchar_255_ resourceId + text title + timestamp_3__with_time_zone updatedAt +} +``` + +--- + +> Generated by [tbls](https://github.com/k1LoW/tbls) diff --git a/docs/generated/postgres-schema/public.instance_ai_threads.md b/docs/generated/postgres-schema/public.instance_ai_threads.md index 8d42516ca55..574683cf297 100644 --- a/docs/generated/postgres-schema/public.instance_ai_threads.md +++ b/docs/generated/postgres-schema/public.instance_ai_threads.md @@ -5,7 +5,7 @@ | Name | Type | Default | Nullable | Children | Parents | Comment | | ---- | ---- | ------- | -------- | -------- | ------- | ------- | | createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | -| id | uuid | | false | [public.ai_builder_temporary_workflow](public.ai_builder_temporary_workflow.md) [public.instance_ai_checkpoints](public.instance_ai_checkpoints.md) [public.instance_ai_iteration_logs](public.instance_ai_iteration_logs.md) [public.instance_ai_messages](public.instance_ai_messages.md) [public.instance_ai_observation_cursors](public.instance_ai_observation_cursors.md) [public.instance_ai_observation_locks](public.instance_ai_observation_locks.md) [public.instance_ai_observational_memory](public.instance_ai_observational_memory.md) [public.instance_ai_observations](public.instance_ai_observations.md) [public.instance_ai_pending_confirmations](public.instance_ai_pending_confirmations.md) [public.instance_ai_run_snapshots](public.instance_ai_run_snapshots.md) [public.instance_ai_thread_grants](public.instance_ai_thread_grants.md) | | | +| id | uuid | | false | [public.ai_builder_temporary_workflow](public.ai_builder_temporary_workflow.md) [public.instance_ai_checkpoints](public.instance_ai_checkpoints.md) [public.instance_ai_events](public.instance_ai_events.md) [public.instance_ai_iteration_logs](public.instance_ai_iteration_logs.md) [public.instance_ai_messages](public.instance_ai_messages.md) [public.instance_ai_observation_cursors](public.instance_ai_observation_cursors.md) [public.instance_ai_observation_locks](public.instance_ai_observation_locks.md) [public.instance_ai_observational_memory](public.instance_ai_observational_memory.md) [public.instance_ai_observations](public.instance_ai_observations.md) [public.instance_ai_pending_confirmations](public.instance_ai_pending_confirmations.md) [public.instance_ai_run_snapshots](public.instance_ai_run_snapshots.md) [public.instance_ai_thread_grants](public.instance_ai_thread_grants.md) | | | | metadata | json | | true | | | | | projectId | varchar(36) | | false | | [public.project](public.project.md) | Project this thread is scoped to | | resourceId | varchar(255) | | false | | | | @@ -40,6 +40,7 @@ erDiagram "public.ai_builder_temporary_workflow" }o--|| "public.instance_ai_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES instance_ai_threads(id) ON DELETE CASCADE" "public.instance_ai_checkpoints" }o--|| "public.instance_ai_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES instance_ai_threads(id) ON DELETE CASCADE" +"public.instance_ai_events" }o--|| "public.instance_ai_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES instance_ai_threads(id) ON DELETE CASCADE" "public.instance_ai_iteration_logs" }o--|| "public.instance_ai_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES instance_ai_threads(id) ON DELETE CASCADE" "public.instance_ai_messages" }o--|| "public.instance_ai_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES instance_ai_threads(id) ON DELETE CASCADE" "public.instance_ai_observation_cursors" |o--|| "public.instance_ai_threads" : "FOREIGN KEY (#quot;observationScopeId#quot;) REFERENCES instance_ai_threads(id) ON DELETE CASCADE" @@ -76,6 +77,15 @@ erDiagram uuid threadId FK timestamp_3__with_time_zone updatedAt } +"public.instance_ai_events" { + timestamp_3__with_time_zone createdAt + text payload + varchar_64_ runId + integer seq + uuid threadId FK + varchar_64_ type + timestamp_3__with_time_zone updatedAt +} "public.instance_ai_iteration_logs" { timestamp_3__with_time_zone createdAt text entry diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md index 15fe8e4a84f..85af9a58838 100644 --- a/docs/generated/sqlite-schema/README.md +++ b/docs/generated/sqlite-schema/README.md @@ -63,6 +63,7 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand. | [installed_nodes](installed_nodes.md) | 4 | | table | | [installed_packages](installed_packages.md) | 6 | | table | | [instance_ai_checkpoints](instance_ai_checkpoints.md) | 8 | | table | +| [instance_ai_events](instance_ai_events.md) | 7 | | table | | [instance_ai_iteration_logs](instance_ai_iteration_logs.md) | 6 | | table | | [instance_ai_mcp_registry_connections](instance_ai_mcp_registry_connections.md) | 7 | | table | | [instance_ai_messages](instance_ai_messages.md) | 8 | | table | @@ -210,6 +211,7 @@ erDiagram "insights_raw" }o--|| "insights_metadata" : "FOREIGN KEY (metaId) REFERENCES insights_metadata (metaId) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "installed_nodes" }o--|| "installed_packages" : "FOREIGN KEY (package) REFERENCES installed_packages (packageName) ON UPDATE CASCADE ON DELETE CASCADE MATCH NONE" "instance_ai_checkpoints" }o--|| "instance_ai_threads" : "FOREIGN KEY (threadId) REFERENCES instance_ai_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" +"instance_ai_events" |o--|| "instance_ai_threads" : "FOREIGN KEY (threadId) REFERENCES instance_ai_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "instance_ai_iteration_logs" }o--|| "instance_ai_threads" : "FOREIGN KEY (threadId) REFERENCES instance_ai_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "instance_ai_mcp_registry_connections" }o--|| "user" : "FOREIGN KEY (userId) REFERENCES user (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "instance_ai_mcp_registry_connections" }o--|| "mcp_registry_server" : "FOREIGN KEY (serverSlug) REFERENCES mcp_registry_server (slug) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" @@ -811,6 +813,15 @@ erDiagram varchar threadId FK datetime_3_ updatedAt } +"instance_ai_events" { + datetime_3_ createdAt + TEXT payload + varchar_64_ runId + INTEGER seq PK + varchar threadId PK + varchar_64_ type + datetime_3_ updatedAt +} "instance_ai_iteration_logs" { datetime_3_ createdAt TEXT entry diff --git a/docs/generated/sqlite-schema/instance_ai_events.md b/docs/generated/sqlite-schema/instance_ai_events.md new file mode 100644 index 00000000000..22fcba71600 --- /dev/null +++ b/docs/generated/sqlite-schema/instance_ai_events.md @@ -0,0 +1,71 @@ +# instance_ai_events + +## Description + +
+Table Definition + +```sql +CREATE TABLE "instance_ai_events" ("threadId" varchar NOT NULL, "seq" integer NOT NULL, "runId" varchar(64) NOT NULL, "type" varchar(64) NOT NULL, "payload" text NOT NULL, "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), CONSTRAINT "FK_35909c5576a4a6c1d6a6fb71caa" FOREIGN KEY ("threadId") REFERENCES "instance_ai_threads" ("id") ON DELETE CASCADE, PRIMARY KEY ("threadId", "seq")) +``` + +
+ +## Columns + +| Name | Type | Default | Nullable | Children | Parents | Comment | +| ---- | ---- | ------- | -------- | -------- | ------- | ------- | +| createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | +| payload | TEXT | | false | | | | +| runId | varchar(64) | | false | | | | +| seq | INTEGER | | false | | | | +| threadId | varchar | | false | | [instance_ai_threads](instance_ai_threads.md) | | +| type | varchar(64) | | false | | | | +| updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | + +## Constraints + +| Name | Type | Definition | +| ---- | ---- | ---------- | +| - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (threadId) REFERENCES instance_ai_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE | +| seq | PRIMARY KEY | PRIMARY KEY (seq) | +| sqlite_autoindex_instance_ai_events_1 | PRIMARY KEY | PRIMARY KEY (threadId, seq) | +| threadId | PRIMARY KEY | PRIMARY KEY (threadId) | + +## Indexes + +| Name | Definition | +| ---- | ---------- | +| IDX_32cdd799675715fb1d2a8683e9 | CREATE INDEX "IDX_32cdd799675715fb1d2a8683e9" ON "instance_ai_events" ("threadId", "runId") | +| sqlite_autoindex_instance_ai_events_1 | PRIMARY KEY (threadId, seq) | + +## Relations + +```mermaid +erDiagram + +"instance_ai_events" |o--|| "instance_ai_threads" : "FOREIGN KEY (threadId) REFERENCES instance_ai_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" + +"instance_ai_events" { + datetime_3_ createdAt + TEXT payload + varchar_64_ runId + INTEGER seq PK + varchar threadId PK + varchar_64_ type + datetime_3_ updatedAt +} +"instance_ai_threads" { + datetime_3_ createdAt + varchar id PK + TEXT metadata + varchar_36_ projectId FK + varchar_255_ resourceId + TEXT title + datetime_3_ updatedAt +} +``` + +--- + +> Generated by [tbls](https://github.com/k1LoW/tbls) diff --git a/docs/generated/sqlite-schema/instance_ai_threads.md b/docs/generated/sqlite-schema/instance_ai_threads.md index 1ccb056cc7e..2fb410ef708 100644 --- a/docs/generated/sqlite-schema/instance_ai_threads.md +++ b/docs/generated/sqlite-schema/instance_ai_threads.md @@ -16,7 +16,7 @@ CREATE TABLE "instance_ai_threads" ("id" varchar PRIMARY KEY NOT NULL, "resource | Name | Type | Default | Nullable | Children | Parents | Comment | | ---- | ---- | ------- | -------- | -------- | ------- | ------- | | createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | -| id | varchar | | false | [ai_builder_temporary_workflow](ai_builder_temporary_workflow.md) [instance_ai_checkpoints](instance_ai_checkpoints.md) [instance_ai_iteration_logs](instance_ai_iteration_logs.md) [instance_ai_messages](instance_ai_messages.md) [instance_ai_observation_cursors](instance_ai_observation_cursors.md) [instance_ai_observation_locks](instance_ai_observation_locks.md) [instance_ai_observational_memory](instance_ai_observational_memory.md) [instance_ai_observations](instance_ai_observations.md) [instance_ai_pending_confirmations](instance_ai_pending_confirmations.md) [instance_ai_run_snapshots](instance_ai_run_snapshots.md) [instance_ai_thread_grants](instance_ai_thread_grants.md) | | | +| id | varchar | | false | [ai_builder_temporary_workflow](ai_builder_temporary_workflow.md) [instance_ai_checkpoints](instance_ai_checkpoints.md) [instance_ai_events](instance_ai_events.md) [instance_ai_iteration_logs](instance_ai_iteration_logs.md) [instance_ai_messages](instance_ai_messages.md) [instance_ai_observation_cursors](instance_ai_observation_cursors.md) [instance_ai_observation_locks](instance_ai_observation_locks.md) [instance_ai_observational_memory](instance_ai_observational_memory.md) [instance_ai_observations](instance_ai_observations.md) [instance_ai_pending_confirmations](instance_ai_pending_confirmations.md) [instance_ai_run_snapshots](instance_ai_run_snapshots.md) [instance_ai_thread_grants](instance_ai_thread_grants.md) | | | | metadata | TEXT | | true | | | | | projectId | varchar(36) | | false | | [project](project.md) | | | resourceId | varchar(255) | | false | | | | @@ -46,6 +46,7 @@ erDiagram "ai_builder_temporary_workflow" }o--|| "instance_ai_threads" : "FOREIGN KEY (threadId) REFERENCES instance_ai_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "instance_ai_checkpoints" }o--|| "instance_ai_threads" : "FOREIGN KEY (threadId) REFERENCES instance_ai_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" +"instance_ai_events" |o--|| "instance_ai_threads" : "FOREIGN KEY (threadId) REFERENCES instance_ai_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "instance_ai_iteration_logs" }o--|| "instance_ai_threads" : "FOREIGN KEY (threadId) REFERENCES instance_ai_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "instance_ai_messages" }o--|| "instance_ai_threads" : "FOREIGN KEY (threadId) REFERENCES instance_ai_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "instance_ai_observation_cursors" |o--|| "instance_ai_threads" : "FOREIGN KEY (observationScopeId) REFERENCES instance_ai_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" @@ -82,6 +83,15 @@ erDiagram varchar threadId FK datetime_3_ updatedAt } +"instance_ai_events" { + datetime_3_ createdAt + TEXT payload + varchar_64_ runId + INTEGER seq PK + varchar threadId PK + varchar_64_ type + datetime_3_ updatedAt +} "instance_ai_iteration_logs" { datetime_3_ createdAt TEXT entry diff --git a/packages/@n8n/db/src/migrations/common/1784000000046-CreateInstanceAiEventsTable.ts b/packages/@n8n/db/src/migrations/common/1784000000046-CreateInstanceAiEventsTable.ts new file mode 100644 index 00000000000..9777f7ec07d --- /dev/null +++ b/packages/@n8n/db/src/migrations/common/1784000000046-CreateInstanceAiEventsTable.ts @@ -0,0 +1,36 @@ +import type { MigrationContext, ReversibleMigration } from '../migration-types'; + +const table = 'instance_ai_events'; + +/** + * Append-only durable event log for Instance AI — the source of truth for + * rendering and SSE replay. Rows are immutable; `seq` is the per-thread + * monotonic replay cursor (assigned in the writer's per-thread drain). + */ +export class CreateInstanceAiEventsTable1784000000046 implements ReversibleMigration { + async up({ schemaBuilder: { createTable, column } }: MigrationContext) { + await createTable(table) + .withColumns( + column('threadId').uuid.primary, + column('seq').int.primary.comment('Per-thread monotonic sequence — the SSE replay cursor'), + column('runId') + .varchar(64) + .notNull.comment('Run that emitted the event — opaque ID from the agent runtime'), + column('type') + .varchar(64) + .notNull.comment('Event type discriminator, duplicated out of the payload'), + column('payload').text.notNull.comment('JSON of the canonical InstanceAiEvent'), + ) + // Run-scoped reads (agent-tree derivation) filter by (threadId, runId). + .withIndexOn(['threadId', 'runId']) + .withForeignKey('threadId', { + tableName: 'instance_ai_threads', + columnName: 'id', + onDelete: 'CASCADE', + }).withTimestamps; + } + + async down({ schemaBuilder: { dropTable } }: MigrationContext) { + await dropTable(table); + } +} diff --git a/packages/@n8n/db/src/migrations/postgresdb/index.ts b/packages/@n8n/db/src/migrations/postgresdb/index.ts index b77178507b8..64b821d97e8 100644 --- a/packages/@n8n/db/src/migrations/postgresdb/index.ts +++ b/packages/@n8n/db/src/migrations/postgresdb/index.ts @@ -219,6 +219,7 @@ import { AddUsedPrivateCredentialsToExecutionEntity1784000000041 } from '../comm import { CreateSchedulerTables1784000000042 } from '../common/1784000000042-CreateSchedulerTables'; import { AddPartialIndexForGlobalCredentials1784000000044 } from '../common/1784000000044-AddPartialIndexForGlobalCredentials'; import { AddRecurringCronScheduleKind1784000000045 } from '../common/1784000000045-AddRecurringCronScheduleKind'; +import { CreateInstanceAiEventsTable1784000000046 } from '../common/1784000000046-CreateInstanceAiEventsTable'; import type { Migration } from '../migration-types'; export const postgresMigrations: Migration[] = [ @@ -443,4 +444,5 @@ export const postgresMigrations: Migration[] = [ CreateWorkflowStatisticsDeltaTable1784000000043, AddPartialIndexForGlobalCredentials1784000000044, AddRecurringCronScheduleKind1784000000045, + CreateInstanceAiEventsTable1784000000046, ]; diff --git a/packages/@n8n/db/src/migrations/sqlite/index.ts b/packages/@n8n/db/src/migrations/sqlite/index.ts index 692865e3bb1..4b6a7d5d62f 100644 --- a/packages/@n8n/db/src/migrations/sqlite/index.ts +++ b/packages/@n8n/db/src/migrations/sqlite/index.ts @@ -211,6 +211,7 @@ import { CreateWorkflowPublicationTriggerStatusTable1784000000040 } from '../com import { AddUsedPrivateCredentialsToExecutionEntity1784000000041 } from '../common/1784000000041-AddUsedPrivateCredentialsToExecutionEntity'; import { CreateSchedulerTables1784000000042 } from '../common/1784000000042-CreateSchedulerTables'; import { AddPartialIndexForGlobalCredentials1784000000044 } from '../common/1784000000044-AddPartialIndexForGlobalCredentials'; +import { CreateInstanceAiEventsTable1784000000046 } from '../common/1784000000046-CreateInstanceAiEventsTable'; const sqliteMigrations: Migration[] = [ InitialMigration1588102412422, @@ -425,6 +426,7 @@ const sqliteMigrations: Migration[] = [ CreateSchedulerTables1784000000042, AddPartialIndexForGlobalCredentials1784000000044, AddRecurringCronScheduleKind1784000000045, + CreateInstanceAiEventsTable1784000000046, ]; export { sqliteMigrations }; diff --git a/packages/cli/src/events/maps/instance-ai.event-map.ts b/packages/cli/src/events/maps/instance-ai.event-map.ts index 4b21bb8b974..0e3cf53e129 100644 --- a/packages/cli/src/events/maps/instance-ai.event-map.ts +++ b/packages/cli/src/events/maps/instance-ai.event-map.ts @@ -1,4 +1,30 @@ export type InstanceAiEventMap = { + /** One durable-log batch persisted by the writer's per-thread drain. */ + 'instance-ai-durable-log-drained': { + rows: number; + bytes: number; + }; + /** publish() enqueue to batch persisted, per published event. */ + 'instance-ai-durable-log-queue-latency': { + ms: number; + }; + /** (threadId, seq) append collision, retried after a reseed (multi-main). */ + 'instance-ai-durable-log-append-conflict': { + attempt: number; + }; + /** A batch dropped after exhausting append retries. */ + 'instance-ai-durable-log-append-failure': { + events: number; + }; + /** An SSE reconnect served a replay from the durable log. */ + 'instance-ai-durable-log-replayed': { + events: number; + cursorAgeEvents: number; + }; + /** History rendered from the message-derived fallback ladder instead of a renderable snapshot tree. */ + 'instance-ai-parser-fallback': { + count: number; + }; 'instance-ai-run-finished': { /** 'suspended' is a non-terminal HITL segment: usage/tool counts only; the terminal event counts the run. */ status: 'completed' | 'cancelled' | 'error' | 'suspended'; diff --git a/packages/cli/src/metrics/prometheus/instance-ai-metrics.service.ts b/packages/cli/src/metrics/prometheus/instance-ai-metrics.service.ts index e4da0bedbdc..399e3d7749d 100644 --- a/packages/cli/src/metrics/prometheus/instance-ai-metrics.service.ts +++ b/packages/cli/src/metrics/prometheus/instance-ai-metrics.service.ts @@ -74,6 +74,77 @@ export class PrometheusInstanceAiMetricsService implements PrometheusMetricsColl }, }); + // Durable event log (RFC: instance-ai durable event log). All series are + // flat when N8N_INSTANCE_AI_DURABLE_LOG is off. + const durableLogRowsTotal = new promClient.Counter({ + name: `${this.config.prefix}instance_ai_durable_log_rows_total`, + help: 'Durable Instance AI event rows appended (structural facts + coalesced blocks).', + }); + durableLogRowsTotal.inc(0); + + const durableLogBytesTotal = new promClient.Counter({ + name: `${this.config.prefix}instance_ai_durable_log_bytes_total`, + help: 'Serialized payload bytes appended to the durable Instance AI event log.', + }); + durableLogBytesTotal.inc(0); + + const durableLogQueueLatency = new promClient.Histogram({ + name: `${this.config.prefix}instance_ai_durable_log_queue_latency_seconds`, + help: 'Time from event publish to durable batch persistence, per event.', + buckets: [0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5], + }); + + const durableLogAppendConflictsTotal = new promClient.Counter({ + name: `${this.config.prefix}instance_ai_durable_log_append_conflicts_total`, + help: 'Retried (threadId, seq) append collisions between concurrent writers.', + }); + durableLogAppendConflictsTotal.inc(0); + + const durableLogAppendFailuresTotal = new promClient.Counter({ + name: `${this.config.prefix}instance_ai_durable_log_append_failures_total`, + help: 'Durable-log batches dropped after exhausting append retries.', + }); + durableLogAppendFailuresTotal.inc(0); + + const durableLogReplaysTotal = new promClient.Counter({ + name: `${this.config.prefix}instance_ai_durable_log_replays_total`, + help: 'SSE reconnects that served a replay from the durable event log.', + }); + durableLogReplaysTotal.inc(0); + + const durableLogReplayCursorAge = new promClient.Histogram({ + name: `${this.config.prefix}instance_ai_durable_log_replay_cursor_age_events`, + help: 'How many events behind the log head a reconnecting cursor was.', + buckets: [1, 5, 25, 100, 500], + }); + + const parserFallbacksTotal = new promClient.Counter({ + name: `${this.config.prefix}instance_ai_parser_fallbacks_total`, + help: 'History messages rendered from the message-derived fallback ladder instead of a renderable snapshot tree.', + }); + parserFallbacksTotal.inc(0); + + this.eventService.on('instance-ai-durable-log-drained', ({ rows, bytes }) => { + durableLogRowsTotal.inc(rows); + durableLogBytesTotal.inc(bytes); + }); + this.eventService.on('instance-ai-durable-log-queue-latency', ({ ms }) => { + durableLogQueueLatency.observe(ms / 1000); + }); + this.eventService.on('instance-ai-durable-log-append-conflict', () => { + durableLogAppendConflictsTotal.inc(1); + }); + this.eventService.on('instance-ai-durable-log-append-failure', () => { + durableLogAppendFailuresTotal.inc(1); + }); + this.eventService.on('instance-ai-durable-log-replayed', ({ cursorAgeEvents }) => { + durableLogReplaysTotal.inc(1); + durableLogReplayCursorAge.observe(cursorAgeEvents); + }); + this.eventService.on('instance-ai-parser-fallback', ({ count }) => { + parserFallbacksTotal.inc(count); + }); + this.eventService.on( 'instance-ai-run-finished', ({ status, durationMs, model, toolCalls, toolErrors, usage }) => { diff --git a/packages/cli/src/modules/instance-ai/entities/index.ts b/packages/cli/src/modules/instance-ai/entities/index.ts index 1f7fe973cbb..437dd33c641 100644 --- a/packages/cli/src/modules/instance-ai/entities/index.ts +++ b/packages/cli/src/modules/instance-ai/entities/index.ts @@ -9,6 +9,7 @@ export { InstanceAiObservationCursor } from './instance-ai-observation-cursor.en export { InstanceAiObservationLock } from './instance-ai-observation-lock.entity'; export { InstanceAiMcpRegistryConnection } from './instance-ai-mcp-registry-connection.entity'; export { InstanceAiThreadGrant } from './instance-ai-thread-grant.entity'; +export { InstanceAiEventLogEntry } from './instance-ai-event-log-entry.entity'; export type { InstanceAiObservationMarker, InstanceAiObservationStatus, diff --git a/packages/cli/src/modules/instance-ai/entities/instance-ai-event-log-entry.entity.ts b/packages/cli/src/modules/instance-ai/entities/instance-ai-event-log-entry.entity.ts new file mode 100644 index 00000000000..ce3602624c4 --- /dev/null +++ b/packages/cli/src/modules/instance-ai/entities/instance-ai-event-log-entry.entity.ts @@ -0,0 +1,42 @@ +import { WithTimestamps } from '@n8n/db'; +import { Column, Entity, Index, JoinColumn, ManyToOne, PrimaryColumn } from '@n8n/typeorm'; + +import { InstanceAiThread } from './instance-ai-thread.entity'; + +/** + * Append-only log of durable Instance AI events — the source of truth for + * rendering and SSE replay (see RFC: instance-ai durable event log). + * + * Rows are immutable: state transitions are new facts appended later (e.g. a + * `run-finish` supersedes in-flight tool calls at fold time), never UPDATEs. + * Token deltas are NOT stored — completed text/reasoning blocks are coalesced + * into one row at the next structural fact. `seq` is the per-thread monotonic + * SSE replay cursor; it survives restarts because it is derived from this + * table (MAX(seq)), unlike the previous in-memory counter. + */ +@Entity({ name: 'instance_ai_events' }) +@Index(['threadId', 'runId']) +export class InstanceAiEventLogEntry extends WithTimestamps { + @ManyToOne(() => InstanceAiThread, { onDelete: 'CASCADE' }) + @JoinColumn({ name: 'threadId' }) + thread: InstanceAiThread; + + @PrimaryColumn({ type: 'uuid' }) + threadId: string; + + /** Per-thread monotonic sequence — the SSE replay cursor. */ + @PrimaryColumn({ type: 'int' }) + seq: number; + + /** Indexed with threadId for run-scoped reads (agent-tree derivation, run summaries). */ + @Column({ type: 'varchar', length: 64 }) + runId: string; + + /** Event type discriminator, duplicated out of the payload for cheap filtering. */ + @Column({ type: 'varchar', length: 64 }) + type: string; + + /** JSON.stringify of the canonical `InstanceAiEvent` (already redacted upstream). */ + @Column({ type: 'text' }) + payload: string; +} diff --git a/packages/cli/src/modules/instance-ai/event-bus/__tests__/durable-event-log.test.ts b/packages/cli/src/modules/instance-ai/event-bus/__tests__/durable-event-log.test.ts new file mode 100644 index 00000000000..35e2ac7bb80 --- /dev/null +++ b/packages/cli/src/modules/instance-ai/event-bus/__tests__/durable-event-log.test.ts @@ -0,0 +1,465 @@ +import type { Logger } from '@n8n/backend-common'; +import type { InstanceAiEvent } from '@n8n/api-types'; +import { QueryFailedError } from '@n8n/typeorm'; +import { mock } from 'vitest-mock-extended'; + +import type { EventService } from '@/events/event.service'; + +import { DurableEventLog, type DrainedEvent } from '../durable-event-log'; +import { DurableLogMetrics } from '../durable-log-metrics'; +import type { InstanceAiEventLogRepository } from '../../repositories/instance-ai-event-log.repository'; + +const THREAD = 'thread-1'; +const RUN = 'run-1'; +const AGENT = 'orchestrator:run-1'; + +/** In-memory stand-in for the repository, with a conflict-injection knob. */ +class FakeRepo { + rows: Array<{ seq: number; event: InstanceAiEvent }> = []; + + /** Simulate a sibling main winning seq ranges: fail the next N appends. */ + failNextAppends = 0; + + /** When an append fails, advance the store as if the sibling wrote rows. */ + siblingRowsPerConflict = 0; + + /** Fail the next N appends with a NON-constraint error (e.g. connectivity). */ + failNextAppendsTransient = 0; + + /** Commit the rows, then fail the response (a lost ack after COMMIT). */ + commitThenFailNextAppends = 0; + + /** Fail the next N maxSeq reads (seq seeding hits a transient DB error). */ + failNextMaxSeq = 0; + + /** Hold the next append until released — models an in-flight DB round trip. */ + gateNextAppend: Promise | undefined; + + async maxSeq(_threadId: string): Promise { + if (this.failNextMaxSeq > 0) { + this.failNextMaxSeq--; + throw new Error('connect ETIMEDOUT'); + } + return this.rows.length ? this.rows[this.rows.length - 1].seq : 0; + } + + async payloadAt(_threadId: string, seq: number): Promise { + const row = this.rows.find((r) => r.seq === seq); + return row ? JSON.stringify(row.event) : null; + } + + async appendBatch(_threadId: string, firstSeq: number, events: InstanceAiEvent[]) { + if (this.gateNextAppend) { + const gate = this.gateNextAppend; + this.gateNextAppend = undefined; + await gate; + } + if (this.failNextAppendsTransient > 0) { + this.failNextAppendsTransient--; + throw new Error('connect ETIMEDOUT'); + } + if (this.commitThenFailNextAppends > 0) { + this.commitThenFailNextAppends--; + events.forEach((event, i) => this.rows.push({ seq: firstSeq + i, event })); + throw new Error('read ECONNRESET'); + } + const conflict = + this.failNextAppends > 0 || + this.rows.some((r) => r.seq >= firstSeq && r.seq < firstSeq + events.length); + if (conflict) { + if (this.failNextAppends > 0) { + this.failNextAppends--; + for (let i = 0; i < this.siblingRowsPerConflict; i++) { + const seq = (await this.maxSeq(_threadId)) + 1; + this.rows.push({ + seq, + event: { + type: 'status', + runId: 'run-sibling', + agentId: 'a', + payload: { message: 'x' }, + }, + }); + } + } + // Same shape a real (threadId, seq) PK collision produces, so the + // writer's isUniqueConstraintError classification is exercised. + throw new QueryFailedError( + 'INSERT INTO instance_ai_events', + [], + Object.assign(new Error('SQLITE_CONSTRAINT: UNIQUE constraint failed: (threadId, seq)'), { + code: 'SQLITE_CONSTRAINT', + }), + ); + } + let bytes = 0; + events.forEach((event, i) => { + bytes += Buffer.byteLength(JSON.stringify(event), 'utf8'); + this.rows.push({ seq: firstSeq + i, event }); + }); + return bytes; + } + + async getAfter(_threadId: string, afterSeq: number) { + return this.rows.filter((r) => r.seq > afterSeq).map((r) => ({ id: r.seq, event: r.event })); + } + + async getForRuns(_threadId: string, runIds: string[]) { + const set = new Set(runIds); + return this.rows.filter((r) => set.has(r.event.runId)).map((r) => r.event); + } +} + +function buildLog(repo: FakeRepo) { + const logger = mock(); + logger.scoped.mockReturnValue(logger); + const metrics = new DurableLogMetrics(mock()); + const log = new DurableEventLog(logger, repo as unknown as InstanceAiEventLogRepository, metrics); + return { log, metrics }; +} + +function textDelta(text: string, responseId = 'msg-1', agentId = AGENT): InstanceAiEvent { + return { type: 'text-delta', runId: RUN, agentId, responseId, payload: { text } }; +} + +function reasoningDelta(text: string, responseId = 'msg-1', agentId = AGENT): InstanceAiEvent { + return { type: 'reasoning-delta', runId: RUN, agentId, responseId, payload: { text } }; +} + +function toolCall(toolCallId: string, agentId = AGENT): InstanceAiEvent { + return { + type: 'tool-call', + runId: RUN, + agentId, + payload: { toolCallId, toolName: 'search-workflows', args: {} }, + }; +} + +function runFinish(): InstanceAiEvent { + return { type: 'run-finish', runId: RUN, agentId: AGENT, payload: { status: 'completed' } }; +} + +/** Publish events through the drain and await it, collecting emissions. */ +async function publishAll( + log: DurableEventLog, + events: InstanceAiEvent[], +): Promise { + const emitted: DrainedEvent[] = []; + for (const event of events) { + log.publish(THREAD, event, (drained) => emitted.push(drained)); + } + await log.flush(THREAD); + return emitted; +} + +describe('DurableEventLog', () => { + it('coalesces a segment into one text-block flushed immediately before the next structural fact', async () => { + const repo = new FakeRepo(); + const { log } = buildLog(repo); + + await publishAll(log, [textDelta('AAA'), textDelta('BBB'), toolCall('tc-1')]); + + const persisted = repo.rows.map((r) => r.event.type); + expect(persisted).toEqual(['text-block', 'tool-call']); + const block = repo.rows[0].event; + expect(block.type === 'text-block' && block.payload.text).toBe('AAABBB'); + expect(block.responseId).toBe('msg-1'); + }); + + it('flushes reasoning and text of one segment as separate blocks, reasoning first', async () => { + const repo = new FakeRepo(); + const { log } = buildLog(repo); + + await publishAll(log, [ + reasoningDelta('think '), + reasoningDelta('hard'), + textDelta('answer'), + toolCall('tc-1'), + ]); + + expect(repo.rows.map((r) => r.event.type)).toEqual([ + 'reasoning-block', + 'text-block', + 'tool-call', + ]); + expect(repo.rows[0].event.payload).toEqual({ text: 'think hard' }); + expect(repo.rows[1].event.payload).toEqual({ text: 'answer' }); + }); + + it('rolls the open segment into a block when the responseId changes (blocks stay 1:1 with segments)', async () => { + const repo = new FakeRepo(); + const { log } = buildLog(repo); + + await publishAll(log, [ + textDelta('first', 'msg-1'), + textDelta(' segment', 'msg-1'), + textDelta('second segment', 'msg-2'), + runFinish(), + ]); + + const blocks = repo.rows.filter((r) => r.event.type === 'text-block'); + expect(blocks).toHaveLength(2); + expect(blocks[0].event.payload).toEqual({ text: 'first segment' }); + expect(blocks[0].event.responseId).toBe('msg-1'); + expect(blocks[1].event.payload).toEqual({ text: 'second segment' }); + expect(blocks[1].event.responseId).toBe('msg-2'); + }); + + it('run-finish flushes the open blocks of every agent in the run', async () => { + const repo = new FakeRepo(); + const { log } = buildLog(repo); + const subAgent = 'sub:run-1:builder'; + + await publishAll(log, [ + textDelta('orchestrator text', 'msg-1', AGENT), + textDelta('sub-agent text', 'msg-s', subAgent), + runFinish(), + ]); + + const blocks = repo.rows.filter((r) => r.event.type === 'text-block'); + expect(blocks.map((b) => b.event.agentId).sort()).toEqual([AGENT, subAgent].sort()); + expect(repo.rows.at(-1)?.event.type).toBe('run-finish'); + }); + + it('live-emits ephemeral events without ids and structural facts with contiguous seqs', async () => { + const repo = new FakeRepo(); + const { log } = buildLog(repo); + + const emitted = await publishAll(log, [ + textDelta('AAA'), + toolCall('tc-1'), + { type: 'status', runId: RUN, agentId: AGENT, payload: { message: 'working' } }, + runFinish(), + ]); + + // The live stream carries every published event, in order. + const live = emitted.filter((e) => e.live); + expect(live.map((e) => e.event.type)).toEqual([ + 'text-delta', + 'tool-call', + 'status', + 'run-finish', + ]); + // Deltas and status carry no id; structural facts carry the DB seq. + expect(live[0].id).toBeUndefined(); + expect(live[2].id).toBeUndefined(); + expect(live[1].id).toBeDefined(); + expect(live[3].id).toBeDefined(); + // Persisted seqs are contiguous from 1. + expect(repo.rows.map((r) => r.seq)).toEqual([1, 2, 3]); + }); + + it('continues the seq across a restart (fresh instance seeds from the DB)', async () => { + const repo = new FakeRepo(); + const { log } = buildLog(repo); + await publishAll(log, [toolCall('tc-1'), toolCall('tc-2')]); + expect(repo.rows.map((r) => r.seq)).toEqual([1, 2]); + + // Restart: a new instance over the same repo. + const { log: log2 } = buildLog(repo); + expect(await log2.getNextEventId(THREAD)).toBe(3); + await publishAll(log2, [toolCall('tc-3')]); + expect(repo.rows.map((r) => r.seq)).toEqual([1, 2, 3]); + }); + + it('retries an append conflict with re-seeded seqs and counts it', async () => { + const repo = new FakeRepo(); + repo.failNextAppends = 1; + repo.siblingRowsPerConflict = 2; // the sibling that won wrote 2 rows + const { log, metrics } = buildLog(repo); + + const emitted = await publishAll(log, [toolCall('tc-1')]); + + expect(metrics.drain.appendConflicts).toBe(1); + expect(metrics.drain.appendFailures).toBe(0); + // Our fact landed after the sibling's rows, with the re-seeded seq. + const ours = repo.rows.find((r) => r.event.type === 'tool-call'); + expect(ours?.seq).toBe(3); + expect(emitted.find((e) => e.live)?.id).toBe(3); + }); + + it('retries a transient append failure without counting it as a conflict', async () => { + const repo = new FakeRepo(); + repo.failNextAppendsTransient = 1; // e.g. a connectivity blip, not a PK collision + const { log, metrics } = buildLog(repo); + + const emitted = await publishAll(log, [toolCall('tc-1')]); + + expect(metrics.drain.appendConflicts).toBe(0); + expect(metrics.drain.appendFailures).toBe(0); + expect(repo.rows.find((r) => r.event.type === 'tool-call')?.seq).toBe(1); + expect(emitted.find((e) => e.live)?.id).toBe(1); + }); + + it('does not duplicate a batch whose append committed but lost its response', async () => { + const repo = new FakeRepo(); + repo.commitThenFailNextAppends = 1; // COMMIT succeeded, the ack was lost + const { log, metrics } = buildLog(repo); + + const emitted = await publishAll(log, [toolCall('tc-1')]); + + // Exactly one row: the retry detected the committed batch instead of + // re-appending it under fresh seqs. + expect(repo.rows.map((r) => [r.seq, r.event.type])).toEqual([[1, 'tool-call']]); + expect(metrics.drain.appendConflicts).toBe(0); + expect(metrics.drain.appendFailures).toBe(0); + expect(metrics.drain.rowsWritten).toBe(1); + expect(emitted.find((e) => e.live)?.id).toBe(1); + }); + + it('retries when seeding the sequence fails instead of rejecting the drain', async () => { + const repo = new FakeRepo(); + repo.failNextMaxSeq = 1; // the currentSeq() seed read hits a transient error + const { log, metrics } = buildLog(repo); + + const emitted = await publishAll(log, [toolCall('tc-1')]); + + expect(metrics.drain.appendConflicts).toBe(0); + expect(repo.rows.map((r) => r.seq)).toEqual([1]); + expect(emitted.find((e) => e.live)?.id).toBe(1); + }); + + it('serializes flush() through the drain so a concurrent fact cannot outrun its block', async () => { + const repo = new FakeRepo(); + const { log } = buildLog(repo); + const emitted: DrainedEvent[] = []; + const collect = (drained: DrainedEvent) => emitted.push(drained); + + log.publish(THREAD, textDelta('streamed tail'), collect); + // A flush (idle timer / shutdown) and a structural fact race: the flush + // marker was queued first, so the block must persist before the fact. + const flushed = log.flush(THREAD); + log.publish(THREAD, toolCall('tc-1'), collect); + await flushed; + await log.flush(THREAD); + + expect(repo.rows.map((r) => [r.seq, r.event.type])).toEqual([ + [1, 'text-block'], + [2, 'tool-call'], + ]); + }); + + it('clearThread during an in-flight append aborts the batch instead of retrying it', async () => { + const repo = new FakeRepo(); + let releaseAppend!: () => void; + repo.gateNextAppend = new Promise((resolve) => (releaseAppend = resolve)); + repo.failNextAppendsTransient = 1; // the gated attempt fails once released + const { log, metrics } = buildLog(repo); + const emitted: DrainedEvent[] = []; + + log.publish(THREAD, toolCall('tc-old'), (drained) => emitted.push(drained)); + // The drain is now awaiting the DB. The thread gets cleared meanwhile + // (deletion / E2E reset); the resumed attempt must stop, not retry into + // the id's next lifecycle. + log.clearThread(THREAD); + releaseAppend(); + await log.flush(THREAD); + + expect(repo.rows).toEqual([]); + expect(emitted).toEqual([]); + expect(metrics.drain.appendConflicts).toBe(0); + // Not a durability incident: the batch was dropped because its thread is gone. + expect(metrics.drain.appendFailures).toBe(0); + }); + + it('a recreated thread id does not receive rows from the previous lifecycle', async () => { + const repo = new FakeRepo(); + let releaseAppend!: () => void; + repo.gateNextAppend = new Promise((resolve) => (releaseAppend = resolve)); + repo.failNextAppendsTransient = 1; + const { log } = buildLog(repo); + const emitted: DrainedEvent[] = []; + const collect = (drained: DrainedEvent) => emitted.push(drained); + + log.publish(THREAD, toolCall('tc-old'), collect); + // Same id, new lifecycle, while the old append is still in flight. + log.clearThread(THREAD); + log.publish(THREAD, toolCall('tc-new'), collect); + releaseAppend(); + await log.flush(THREAD); + + // Only the new lifecycle's fact exists, from seq 1. + const persisted = repo.rows.map((r) => [ + r.seq, + r.event.type === 'tool-call' ? r.event.payload.toolCallId : r.event.type, + ]); + expect(persisted).toEqual([[1, 'tc-new']]); + expect(emitted.filter((e) => e.live).map((e) => e.id)).toEqual([1]); + }); + + it('getNextEventId reflects rows appended by another writer', async () => { + const repo = new FakeRepo(); + const { log } = buildLog(repo); + await publishAll(log, [toolCall('tc-1'), toolCall('tc-2')]); + + // A sibling main appends directly to the shared table. + repo.rows.push({ + seq: 3, + event: { type: 'status', runId: 'run-sibling', agentId: 'a', payload: { message: 'x' } }, + }); + + // The cursor authority is the DB, not this main's last-append cache. + expect(await log.getNextEventId(THREAD)).toBe(4); + }); + + it('drops the batch after exhausting retries but still delivers live', async () => { + const repo = new FakeRepo(); + repo.failNextAppends = 99; // never succeeds + const { log, metrics } = buildLog(repo); + + const emitted = await publishAll(log, [toolCall('tc-1')]); + + expect(metrics.drain.appendFailures).toBe(1); + expect(repo.rows.filter((r) => r.event.type === 'tool-call')).toHaveLength(0); + const live = emitted.filter((e) => e.live); + expect(live).toHaveLength(1); + expect(live[0].id).toBeUndefined(); // nothing durable to point a cursor at + }); + + it('flush() persists still-open buffers as blocks (shutdown path)', async () => { + const repo = new FakeRepo(); + const { log } = buildLog(repo); + + // Open segment with no structural fact after it: only flush() closes it. + await publishAll(log, [textDelta('tail of a streamed answer')]); + + const blocks = repo.rows.filter((r) => r.event.type === 'text-block'); + expect(blocks).toHaveLength(1); + expect(blocks[0].event.payload).toEqual({ text: 'tail of a streamed answer' }); + }); + it('clearThread drops the stale seq cache so a reused thread reseeds from the DB', async () => { + const repo = new FakeRepo(); + const { log } = buildLog(repo); + await publishAll(log, [toolCall('tc-1'), toolCall('tc-2')]); + expect(repo.rows.map((r) => r.seq)).toEqual([1, 2]); + + // Thread deleted: rows cascade away, the log's per-thread state clears. + repo.rows = []; + log.clearThread(THREAD); + + // A later publish for the (re)created thread starts clean at seq 1, + // with no conflict retries against the stale cached counter. + await publishAll(log, [toolCall('tc-3')]); + expect(repo.rows.map((r) => r.seq)).toEqual([1]); + }); + it('idle flush persists trailing deltas that no structural fact ever follows', async () => { + // e.g. a terminal-outcome line published after run-finish, or a liveness + // timeout notice: without the idle flush these would never reach the log + // and disappear on reload. + const repo = new FakeRepo(); + const { log } = buildLog(repo); + log.idleFlushMs = 25; + + const emitted = []; + log.publish(THREAD, textDelta('The background workflow-builder task was cancelled.'), (d) => + emitted.push(d), + ); + await new Promise((resolve) => setTimeout(resolve, 120)); + + const blocks = repo.rows.filter((r) => r.event.type === 'text-block'); + expect(blocks).toHaveLength(1); + expect(blocks[0].event.payload).toEqual({ + text: 'The background workflow-builder task was cancelled.', + }); + }); +}); diff --git a/packages/cli/src/modules/instance-ai/event-bus/durable-event-log.ts b/packages/cli/src/modules/instance-ai/event-bus/durable-event-log.ts new file mode 100644 index 00000000000..ff9cf6cf2ba --- /dev/null +++ b/packages/cli/src/modules/instance-ai/event-bus/durable-event-log.ts @@ -0,0 +1,618 @@ +import { INSTANCE_AI_EPHEMERAL_EVENT_TYPES, type InstanceAiEvent } from '@n8n/api-types'; +import { Logger } from '@n8n/backend-common'; +import { isUniqueConstraintError } from '@n8n/db'; +import { Service } from '@n8n/di'; +import type { StoredEvent } from '@n8n/instance-ai'; + +import { DurableLogMetrics } from './durable-log-metrics'; +import { InstanceAiEventLogRepository } from '../repositories/instance-ai-event-log.repository'; + +/** + * Rule: streaming granularity ≠ persistence granularity. Deltas are transport; + * steps are state (see RFC: instance-ai durable event log). + * + * Every published event is exactly one of: + * - EPHEMERAL — live-emitted with NO seq (SSE frame without `id:`, so the + * browser replay cursor never points at it); never persisted. Text/reasoning + * deltas are additionally buffered for coalescing. + * - COALESCED — the buffered deltas of a completed block, flushed as ONE + * persisted row immediately before the next structural fact. Never + * live-emitted (live clients already saw the deltas). + * - STRUCTURAL — persisted with a seq AND live-emitted. These are the facts + * the fold reconstructs state from. + * + * Live stream = ephemeral + structural. Replay stream = coalesced + structural. + * Both are complete; a block always flushes before the fact that follows it, + * so any cursor taken from a structural fact replays exactly the missing tail. + * + * The ephemeral list is shared with the frontend's SSE dedup gate + * (INSTANCE_AI_EPHEMERAL_EVENT_TYPES, #33915) so the two sides cannot drift. + */ +const EPHEMERAL_TYPES = INSTANCE_AI_EPHEMERAL_EVENT_TYPES; + +/** Retries per batch on (threadId, seq) PK collision before giving up. */ +const MAX_APPEND_ATTEMPTS = 5; + +/** + * Trailing deltas with no structural fact after them (e.g. a terminal-outcome + * line published after run-finish, or a liveness timeout notice) would sit in + * the coalescer forever; after this quiet period the open buffers are flushed + * as blocks so they reach the durable log and replay/history stay complete. + */ +const IDLE_FLUSH_MS = 3_000; + +/** How a drained event reaches the bus. `id` set = durable; `live` = emit to SSE. */ +export interface DrainedEvent { + id?: number; + event: InstanceAiEvent; + live: boolean; +} + +interface PendingEvent { + event: InstanceAiEvent; + enqueuedAt: number; +} + +interface FlushSignal { + resolve: () => void; +} + +/** + * A flush marker rides the same per-thread queue as events, so open buffers + * are persisted exactly at the marker's position: everything published before + * it lands first, everything after it lands later. Persisting outside the + * drain would race a concurrent publish for seqs and could reorder a block + * after the structural fact it must precede. + */ +type PendingEntry = PendingEvent | { flushSignal: FlushSignal }; + +function isFlushMarker(entry: PendingEntry): entry is { flushSignal: FlushSignal } { + return 'flushSignal' in entry; +} + +function serializedBytes(events: InstanceAiEvent[]): number { + return events.reduce( + (total, event) => total + Buffer.byteLength(JSON.stringify(event), 'utf8'), + 0, + ); +} + +interface CoalesceBuffer { + text: string[]; + reasoning: string[]; + /** responseId of the segment being coalesced — carried on the flushed block + * so the reducer can REPLACE the segment's streamed deltas on replay. */ + textResponseId?: string; + reasoningResponseId?: string; +} + +type EmitFn = (drained: DrainedEvent) => void; + +@Service() +export class DurableEventLog { + /** publish() stays synchronous: events queue here, a per-thread drain assigns seq. */ + private readonly pendingByThread = new Map(); + + /** In-flight drain per thread, awaited by flush(). */ + private readonly draining = new Map>(); + + /** Last assigned seq per thread; lazily seeded from MAX(seq) in the DB. */ + private readonly lastSeq = new Map(); + + /** Open text/reasoning blocks per thread, keyed `${runId}:${agentId}`. */ + private readonly buffers = new Map>(); + + private readonly emitters = new Map(); + + /** + * Lifecycle token per thread, compared by identity. A drain captures it at + * batch start and re-checks after every await: clearThread() replaces the + * thread's token, so a drain resuming from a DB round trip after the clear + * aborts instead of persisting or emitting into the id's next lifecycle + * (e.g. a thread deleted and recreated under the same id mid-append). + */ + private readonly lifecycles = new Map(); + + /** Per-thread idle timers driving the trailing-delta flush. */ + private readonly idleFlushTimers = new Map(); + + /** Overridable for tests. */ + idleFlushMs = IDLE_FLUSH_MS; + + constructor( + private readonly logger: Logger, + private readonly repo: InstanceAiEventLogRepository, + private readonly metrics: DurableLogMetrics, + ) { + this.logger = this.logger.scoped('instance-ai'); + } + + /** Synchronous enqueue — ordering is preserved by the single per-thread drain. */ + publish(threadId: string, event: InstanceAiEvent, emit: EmitFn): void { + this.emitters.set(threadId, emit); + const pending = this.pendingByThread.get(threadId); + const entry: PendingEvent = { event, enqueuedAt: Date.now() }; + if (pending) pending.push(entry); + else this.pendingByThread.set(threadId, [entry]); + this.ensureDraining(threadId); + this.scheduleIdleFlush(threadId); + } + + /** (Re)arm the trailing-delta flush: fires only when the thread goes quiet. */ + private scheduleIdleFlush(threadId: string): void { + const existing = this.idleFlushTimers.get(threadId); + if (existing) clearTimeout(existing); + const timer = setTimeout(() => { + this.idleFlushTimers.delete(threadId); + void this.flush(threadId).catch((error) => { + this.logger.error('Instance AI event log idle flush failed', { threadId, error }); + }); + }, this.idleFlushMs); + timer.unref(); + this.idleFlushTimers.set(threadId, timer); + } + + async getEventsAfter(threadId: string, afterSeq: number): Promise { + return await this.repo.getAfter(threadId, afterSeq); + } + + async getEventsForRuns(threadId: string, runIds: string[]): Promise { + return await this.repo.getForRuns(threadId, runIds); + } + + async getNextEventId(threadId: string): Promise { + // Always from the DB: the local cache only tracks THIS main's appends and + // goes stale the moment a sibling main wins a seq range, which would seed + // the client's replay cursor below rows its history response already + // covered. The cache stays writer-only (persistWithRetry). + return (await this.repo.maxSeq(threadId)) + 1; + } + + /** + * Drop per-thread drain state (seq cache, pending queue, open buffers, + * emitter). Called when a thread is cleared or deleted: a straggler publish + * after deletion would otherwise append against a stale seq cache and burn + * the retry loop on the thread FK, and the per-thread maps would grow + * unbounded across a long-lived process. + */ + clearThread(threadId: string): void { + this.resolvePendingFlushes(threadId); + this.pendingByThread.delete(threadId); + this.lastSeq.delete(threadId); + this.buffers.delete(threadId); + this.emitters.delete(threadId); + // Invalidates any drain currently awaiting the DB for this thread: it + // re-checks the token when it resumes and aborts its persist/emits. + this.lifecycles.delete(threadId); + const timer = this.idleFlushTimers.get(threadId); + if (timer) clearTimeout(timer); + this.idleFlushTimers.delete(threadId); + } + + /** Drop all per-thread drain state. Used during module shutdown. */ + clear(): void { + for (const threadId of this.pendingByThread.keys()) this.resolvePendingFlushes(threadId); + this.pendingByThread.clear(); + this.lastSeq.clear(); + this.buffers.clear(); + this.emitters.clear(); + this.lifecycles.clear(); + for (const timer of this.idleFlushTimers.values()) clearTimeout(timer); + this.idleFlushTimers.clear(); + } + + /** A cleared thread has nothing left to flush — settle waiters so they never hang. */ + private resolvePendingFlushes(threadId: string): void { + const pending = this.pendingByThread.get(threadId); + if (!pending) return; + for (const entry of pending) { + if (isFlushMarker(entry)) entry.flushSignal.resolve(); + } + } + + /** + * Persist any still-open coalesce buffers as blocks so streamed text + * survives a shutdown mid-segment. Serialized through the per-thread drain + * (as a queue marker), so it can never race a concurrent publish for seqs. + */ + async flush(threadId: string): Promise { + const hasBuffers = (this.buffers.get(threadId)?.size ?? 0) > 0; + if (!this.pendingByThread.has(threadId) && !this.draining.has(threadId) && !hasBuffers) { + return; + } + await new Promise((resolve) => { + const entry: PendingEntry = { flushSignal: { resolve } }; + const pending = this.pendingByThread.get(threadId); + if (pending) pending.push(entry); + else this.pendingByThread.set(threadId, [entry]); + this.ensureDraining(threadId); + }); + } + + /** Drain shutdown flush: called from module shutdown so no thread loses its tail. */ + async flushAll(): Promise { + const threadIds = new Set([ + ...this.draining.keys(), + ...this.buffers.keys(), + ...this.pendingByThread.keys(), + ]); + for (const threadId of threadIds) { + try { + await this.flush(threadId); + } catch (error) { + this.logger.error('Failed to flush Instance AI event log on shutdown', { + threadId, + error, + }); + } + } + } + + private ensureDraining(threadId: string): void { + if (this.draining.has(threadId)) return; + const drain = (async () => { + try { + let batch = this.takePending(threadId); + while (batch.length > 0) { + try { + await this.drainBatch(threadId, batch); + } catch (error) { + // Keep the drain alive: a failed batch must not reject the + // (unawaited) drain promise or strand later publishes/flushes. + this.logger.error('Instance AI event log drain failed for a batch', { + threadId, + error, + }); + for (const entry of batch) { + if (isFlushMarker(entry)) entry.flushSignal.resolve(); + } + } + batch = this.takePending(threadId); + } + } finally { + this.draining.delete(threadId); + } + })(); + this.draining.set(threadId, drain); + } + + private async drainBatch(threadId: string, batch: PendingEntry[]): Promise { + const lifecycle = this.currentLifecycle(threadId); + const flushSignals: FlushSignal[] = []; + const settleFlushes = () => { + for (const signal of flushSignals) signal.resolve(); + }; + + const emit = this.emitters.get(threadId); + if (!emit) { + for (const entry of batch) { + if (isFlushMarker(entry)) entry.flushSignal.resolve(); + } + return; + } + + // Build the batch plan first; seqs are assigned inside persistWithRetry so + // an append conflict can re-assign them from a re-seeded counter. + const toPersist: InstanceAiEvent[] = []; + const toEmit: Array<{ event: InstanceAiEvent; persistIndex?: number; live: boolean }> = []; + + for (const entry of batch) { + if (isFlushMarker(entry)) { + // Persist every open buffer at the marker's queue position, so the + // flush is ordered exactly against the events published around it. + for (const block of this.takeAllOpenBlocks(threadId)) { + toEmit.push({ event: block, persistIndex: toPersist.length, live: false }); + toPersist.push(block); + } + flushSignals.push(entry.flushSignal); + continue; + } + const { event } = entry; + if (EPHEMERAL_TYPES.has(event.type)) { + // A delta with a new responseId starts a new segment: close the old + // one as a block first, so blocks stay exactly 1:1 with segments and + // the reducer's open-segment replacement stays exact. + for (const rolled of this.rollSegmentOnResponseChange(threadId, event)) { + toEmit.push({ event: rolled, persistIndex: toPersist.length, live: false }); + toPersist.push(rolled); + } + this.bufferDelta(threadId, event); + toEmit.push({ event, live: true }); + continue; + } + // Structural fact: flush this agent's open blocks first so replay + // order matches live order (block content precedes the fact). + for (const block of this.flushBlocks(threadId, event)) { + toEmit.push({ event: block, persistIndex: toPersist.length, live: false }); + toPersist.push(block); + } + toEmit.push({ event, persistIndex: toPersist.length, live: true }); + toPersist.push(event); + } + + let firstSeq: number | undefined; + if (toPersist.length > 0) { + firstSeq = await this.persistWithRetry(threadId, toPersist, lifecycle); + // The thread was cleared while the persist was in flight: its next + // lifecycle (a recreated id, or nothing) must not receive this batch's + // emissions. Flush waiters still settle — there is nothing left to flush. + if (this.lifecycles.get(threadId) !== lifecycle) { + settleFlushes(); + return; + } + const persistedAt = Date.now(); + for (const entry of batch) { + if (!isFlushMarker(entry)) this.metrics.recordQueueLatency(persistedAt - entry.enqueuedAt); + } + } + + for (const drained of toEmit) { + const id = + drained.persistIndex !== undefined && firstSeq !== undefined + ? firstSeq + drained.persistIndex + : undefined; + emit({ ...(id !== undefined ? { id } : {}), event: drained.event, live: drained.live }); + } + settleFlushes(); + } + + /** The thread's current lifecycle token, minted on first use. */ + private currentLifecycle(threadId: string): object { + let token = this.lifecycles.get(threadId); + if (!token) { + token = {}; + this.lifecycles.set(threadId, token); + } + return token; + } + + /** Drain every open buffer of the thread into block facts (flush marker path). */ + private takeAllOpenBlocks(threadId: string): InstanceAiEvent[] { + const threadBuffers = this.buffers.get(threadId); + if (!threadBuffers || threadBuffers.size === 0) return []; + + const blocks: InstanceAiEvent[] = []; + for (const [key, buffer] of threadBuffers) { + const separator = key.indexOf(':'); + const runId = key.slice(0, separator); + const agentId = key.slice(separator + 1); + const reasoning = this.takeBlock('reasoning', runId, agentId, buffer); + if (reasoning) blocks.push(reasoning); + const text = this.takeBlock('text', runId, agentId, buffer); + if (text) blocks.push(text); + } + this.buffers.delete(threadId); + return blocks; + } + + /** + * Append `events` with contiguous seqs, retrying on (threadId, seq) PK + * collision — another main won the range (multi-main only), so re-seed from + * the DB and try again. Returns the first assigned seq, or undefined when + * the batch had to be dropped (logged; live delivery still happens). + * INS-844 merges the shared-sequence drain (#33558) here: its Redis INCRBY + * becomes this batch-INSERT's id assignment, one round trip. + */ + private async persistWithRetry( + threadId: string, + events: InstanceAiEvent[], + lifecycle: object, + ): Promise { + let lastError: unknown; + for (let attempt = 1; attempt <= MAX_APPEND_ATTEMPTS; attempt++) { + // The thread was cleared while an earlier attempt was in flight: stop + // instead of appending into the id's next lifecycle (deleted threads + // would burn every retry on the FK; a recreated id would accept the + // stale rows). Not a durability failure — the thread is gone. + if (this.lifecycles.get(threadId) !== lifecycle) { + this.logger.debug('Instance AI event log dropped a batch for a cleared thread', { + threadId, + events: events.length, + }); + return undefined; + } + // The seed read lives inside the try: a transient failure there must + // consume an attempt and retry, not reject the (unawaited) drain. + let firstSeq: number | undefined; + try { + firstSeq = (await this.currentSeq(threadId)) + 1; + const bytes = await this.repo.appendBatch(threadId, firstSeq, events); + this.lastSeq.set(threadId, firstSeq + events.length - 1); + this.metrics.recordDrainBatch(events.length, bytes); + return firstSeq; + } catch (error) { + lastError = error; + if (isUniqueConstraintError(error)) { + // (threadId, seq) collision: another main won the range — re-seed + // from the DB and try again. + this.lastSeq.delete(threadId); + this.metrics.recordAppendConflict(attempt); + this.logger.warn('Instance AI event log append conflict, retrying', { + threadId, + attempt, + error, + }); + continue; + } + // Transient failure (seed read, connectivity, timeout). The append is + // a single INSERT, so if its commit outran a lost response the batch + // is already durable — detect that instead of re-appending it under + // fresh seqs, which would duplicate every fact in the replay. + if (firstSeq !== undefined && (await this.didBatchCommit(threadId, firstSeq, events))) { + this.lastSeq.set(threadId, firstSeq + events.length - 1); + this.metrics.recordDrainBatch(events.length, serializedBytes(events)); + return firstSeq; + } + // Also covers a PK violation a driver reports under a code the + // detector doesn't know: the committed row differs from ours, so the + // re-seed below realigns and the next attempt lands cleanly. + this.lastSeq.delete(threadId); + this.logger.warn('Instance AI event log append failed, retrying', { + threadId, + attempt, + error, + }); + } + } + this.metrics.recordAppendFailure(events.length); + this.logger.error('Instance AI event log append failed, dropping batch', { + threadId, + events: events.length, + error: lastError, + }); + return undefined; + } + + /** + * Whether a batch whose append errored actually committed: the first row of + * the attempted range exists with exactly our payload (single-statement + * INSERT, so the first row proves the whole batch). A read failure means + * the DB is still unreachable — report not-committed and keep retrying. + */ + private async didBatchCommit( + threadId: string, + firstSeq: number, + events: InstanceAiEvent[], + ): Promise { + try { + const payload = await this.repo.payloadAt(threadId, firstSeq); + return payload !== null && payload === JSON.stringify(events[0]); + } catch { + return false; + } + } + + /** Append a delta to its agent's open block. */ + private bufferDelta(threadId: string, event: InstanceAiEvent): void { + if (event.type !== 'text-delta' && event.type !== 'reasoning-delta') return; + if (!event.responseId) { + // Every delta producer stamps a responseId (provider-supplied or a + // synthetic segment id); an id-less delta would flush an id-less block + // and break the reducer's segment-keyed replace semantics on replay. + this.logger.debug('Instance AI durable log buffered a delta without a responseId', { + threadId, + runId: event.runId, + agentId: event.agentId, + }); + } + const buffer = this.getOrCreateBuffer(threadId, `${event.runId}:${event.agentId}`); + if (event.type === 'text-delta') { + buffer.text.push(event.payload.text); + buffer.textResponseId = event.responseId ?? buffer.textResponseId; + } else { + buffer.reasoning.push(event.payload.text); + buffer.reasoningResponseId = event.responseId ?? buffer.reasoningResponseId; + } + } + + /** + * Close open blocks made stale by a structural fact: the fact's own agent + * always; every agent of the run on `run-finish`. Blocks are persisted as + * `text-block`/`reasoning-block` facts carrying the segment's responseId — + * on replay the reducer REPLACES the segment's streamed deltas, so a client + * that reconnects mid-block never sees partial text or reasoning twice. + */ + private flushBlocks(threadId: string, fact: InstanceAiEvent): InstanceAiEvent[] { + const threadBuffers = this.buffers.get(threadId); + if (!threadBuffers) return []; + const keys = + fact.type === 'run-finish' + ? [...threadBuffers.keys()].filter((k) => k.startsWith(`${fact.runId}:`)) + : [`${fact.runId}:${fact.agentId}`]; + + const flushed: InstanceAiEvent[] = []; + for (const key of keys) { + const buffer = threadBuffers.get(key); + if (!buffer) continue; + threadBuffers.delete(key); + const agentId = key.slice(fact.runId.length + 1); + const reasoning = this.takeBlock('reasoning', fact.runId, agentId, buffer); + if (reasoning) flushed.push(reasoning); + const text = this.takeBlock('text', fact.runId, agentId, buffer); + if (text) flushed.push(text); + } + return flushed; + } + + /** + * A delta whose responseId differs from its kind's open buffer starts a new + * segment (e.g. consecutive reasoning-only steps with no tool call between). + * Close the previous segment as a block so blocks stay 1:1 with segments. + */ + private rollSegmentOnResponseChange(threadId: string, event: InstanceAiEvent): InstanceAiEvent[] { + if (event.type !== 'text-delta' && event.type !== 'reasoning-delta') return []; + const buffer = this.buffers.get(threadId)?.get(`${event.runId}:${event.agentId}`); + if (!buffer) return []; + const kind = event.type === 'text-delta' ? 'text' : 'reasoning'; + const openResponseId = kind === 'text' ? buffer.textResponseId : buffer.reasoningResponseId; + const hasContent = (kind === 'text' ? buffer.text : buffer.reasoning).length > 0; + if (!hasContent || openResponseId === event.responseId) return []; + const block = this.takeBlock(kind, event.runId, event.agentId, buffer); + return block ? [block] : []; + } + + /** Drain one kind's open segment from a buffer into its block fact. */ + private takeBlock( + kind: 'text' | 'reasoning', + runId: string, + agentId: string, + buffer: CoalesceBuffer, + ): InstanceAiEvent | undefined { + const parts = kind === 'text' ? buffer.text : buffer.reasoning; + if (parts.length === 0) return undefined; + const text = parts.join(''); + if (kind === 'text') { + const responseId = buffer.textResponseId; + buffer.text = []; + buffer.textResponseId = undefined; + return { + type: 'text-block', + runId, + agentId, + ...(responseId ? { responseId } : {}), + payload: { text }, + }; + } + const responseId = buffer.reasoningResponseId; + buffer.reasoning = []; + buffer.reasoningResponseId = undefined; + return { + type: 'reasoning-block', + runId, + agentId, + ...(responseId ? { responseId } : {}), + payload: { text }, + }; + } + + private async currentSeq(threadId: string): Promise { + const cached = this.lastSeq.get(threadId); + if (cached !== undefined) return cached; + const max = await this.repo.maxSeq(threadId); + // Cutover note (RFC Q&A on cursors): INS-844 seeds from max(DB, Redis + // high-water mark) so cursors minted by the live shared sequence stay valid. + this.lastSeq.set(threadId, max); + return max; + } + + private getOrCreateBuffer(threadId: string, key: string): CoalesceBuffer { + let threadBuffers = this.buffers.get(threadId); + if (!threadBuffers) { + threadBuffers = new Map(); + this.buffers.set(threadId, threadBuffers); + } + let buffer = threadBuffers.get(key); + if (!buffer) { + buffer = { text: [], reasoning: [] }; + threadBuffers.set(key, buffer); + } + return buffer; + } + + private takePending(threadId: string): PendingEntry[] { + const pending = this.pendingByThread.get(threadId); + if (!pending) return []; + this.pendingByThread.delete(threadId); + return pending; + } +} diff --git a/packages/cli/src/modules/instance-ai/event-bus/durable-log-metrics.ts b/packages/cli/src/modules/instance-ai/event-bus/durable-log-metrics.ts new file mode 100644 index 00000000000..505394b0580 --- /dev/null +++ b/packages/cli/src/modules/instance-ai/event-bus/durable-log-metrics.ts @@ -0,0 +1,90 @@ +import { Service } from '@n8n/di'; + +import { EventService } from '@/events/event.service'; + +/** + * Durable-log instrumentation (RFC: instance-ai durable event log, + * "Instrumentation"). Two consumers, one recording point: + * + * - Typed `EventService` events feed the Prometheus collectors + * (`PrometheusInstanceAiMetricsService`), following the same convention as + * `instance-ai-run-finished`. + * - The in-process counters below are kept in lockstep for synchronous reads + * in unit tests and log lines. + */ +@Service() +export class DurableLogMetrics { + /** Writer-side counters, recorded inside DurableEventLog's per-thread drain. */ + drain = { + batches: 0, + /** Durable rows written (structural facts + coalesced blocks). */ + rowsWritten: 0, + /** Serialized payload bytes written. */ + bytesWritten: 0, + /** (threadId, seq) PK collisions — another main won the range. */ + appendConflicts: 0, + /** Batches dropped after exhausting append retries. */ + appendFailures: 0, + /** publish() enqueue → batch persisted, per event. */ + queueLatencyMsTotal: 0, + queueLatencyMsMax: 0, + queueLatencySamples: 0, + }; + + /** SSE endpoint counters (replay path). */ + sse = { + /** Reconnects that served a replay from the durable log. */ + replaysServed: 0, + /** Events delivered by those replays. */ + replayEventsServed: 0, + /** Cursor age at replay time, in events behind the log head. */ + cursorAgeEventsTotal: 0, + cursorAgeEventsMax: 0, + }; + + constructor(private readonly eventService: EventService) {} + + recordDrainBatch(rows: number, bytes: number): void { + this.drain.batches++; + this.drain.rowsWritten += rows; + this.drain.bytesWritten += bytes; + this.eventService.emit('instance-ai-durable-log-drained', { rows, bytes }); + } + + recordQueueLatency(ms: number): void { + this.drain.queueLatencyMsTotal += ms; + this.drain.queueLatencyMsMax = Math.max(this.drain.queueLatencyMsMax, ms); + this.drain.queueLatencySamples++; + this.eventService.emit('instance-ai-durable-log-queue-latency', { ms }); + } + + recordAppendConflict(attempt: number): void { + this.drain.appendConflicts++; + this.eventService.emit('instance-ai-durable-log-append-conflict', { attempt }); + } + + recordAppendFailure(events: number): void { + this.drain.appendFailures++; + this.eventService.emit('instance-ai-durable-log-append-failure', { events }); + } + + recordReplay(eventsServed: number, cursorAgeEvents: number): void { + this.sse.replaysServed++; + this.sse.replayEventsServed += eventsServed; + this.sse.cursorAgeEventsTotal += cursorAgeEvents; + this.sse.cursorAgeEventsMax = Math.max(this.sse.cursorAgeEventsMax, cursorAgeEvents); + this.eventService.emit('instance-ai-durable-log-replayed', { + events: eventsServed, + cursorAgeEvents, + }); + } + + /** + * The parser is pure module code and keeps its own counter + * (messageParserStats); this only forwards new activations to the + * metrics pipeline. + */ + notifyParserFallbacks(count: number): void { + if (count > 0) this.eventService.emit('instance-ai-parser-fallback', { count }); + } +} diff --git a/packages/cli/src/modules/instance-ai/instance-ai.module.ts b/packages/cli/src/modules/instance-ai/instance-ai.module.ts index 3db54835838..b9edb5ce0d9 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.module.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.module.ts @@ -76,6 +76,9 @@ export class InstanceAiModule implements ModuleInterface { './entities/instance-ai-mcp-registry-connection.entity' ); const { InstanceAiThreadGrant } = await import('./entities/instance-ai-thread-grant.entity'); + const { InstanceAiEventLogEntry } = await import( + './entities/instance-ai-event-log-entry.entity' + ); return [ InstanceAiThread, @@ -90,6 +93,7 @@ export class InstanceAiModule implements ModuleInterface { InstanceAiObservationLock, InstanceAiMcpRegistryConnection, InstanceAiThreadGrant, + InstanceAiEventLogEntry, ]; } diff --git a/packages/cli/src/modules/instance-ai/repositories/index.ts b/packages/cli/src/modules/instance-ai/repositories/index.ts index 7498b78e21d..c62999775ea 100644 --- a/packages/cli/src/modules/instance-ai/repositories/index.ts +++ b/packages/cli/src/modules/instance-ai/repositories/index.ts @@ -9,3 +9,4 @@ export { InstanceAiObservationCursorRepository } from './instance-ai-observation export { InstanceAiObservationLockRepository } from './instance-ai-observation-lock.repository'; export { InstanceAiMcpRegistryConnectionRepository } from './instance-ai-mcp-registry-connection.repository'; export { InstanceAiThreadGrantRepository } from './instance-ai-thread-grant.repository'; +export { InstanceAiEventLogRepository } from './instance-ai-event-log.repository'; diff --git a/packages/cli/src/modules/instance-ai/repositories/instance-ai-event-log.repository.ts b/packages/cli/src/modules/instance-ai/repositories/instance-ai-event-log.repository.ts new file mode 100644 index 00000000000..c58a20fb12d --- /dev/null +++ b/packages/cli/src/modules/instance-ai/repositories/instance-ai-event-log.repository.ts @@ -0,0 +1,80 @@ +import type { InstanceAiEvent } from '@n8n/api-types'; +import { Service } from '@n8n/di'; +import type { StoredEvent } from '@n8n/instance-ai'; +import { DataSource, MoreThan, Repository } from '@n8n/typeorm'; +import { jsonParse } from 'n8n-workflow'; + +import { InstanceAiEventLogEntry } from '../entities/instance-ai-event-log-entry.entity'; + +@Service() +export class InstanceAiEventLogRepository extends Repository { + constructor(dataSource: DataSource) { + super(InstanceAiEventLogEntry, dataSource.manager); + } + + /** Highest seq assigned for a thread, 0 when the log is empty. */ + async maxSeq(threadId: string): Promise { + const row = await this.createQueryBuilder('e') + .select('MAX(e.seq)', 'max') + .where('e.threadId = :threadId', { threadId }) + .getRawOne<{ max: number | null }>(); + return row?.max ?? 0; + } + + /** + * Append a batch of events with contiguous seq values starting at `firstSeq`, + * in one transaction. The (threadId, seq) PK makes a concurrent-writer race + * fail loudly instead of silently interleaving — the caller re-reads maxSeq + * and retries. Returns the serialized payload bytes written (instrumentation). + * + * INS-844 (compose with the shared-sequence drain): the live-id Redis INCRBY + * merges into this call, so id assignment and durable insert become one round trip. + */ + async appendBatch( + threadId: string, + firstSeq: number, + events: InstanceAiEvent[], + ): Promise { + let bytes = 0; + const rows = events.map((event, i) => { + const payload = JSON.stringify(event); + bytes += Buffer.byteLength(payload, 'utf8'); + return { + threadId, + seq: firstSeq + i, + runId: event.runId, + type: event.type, + payload, + }; + }); + await this.insert(rows); + return bytes; + } + + /** + * Raw payload at an exact (threadId, seq), or null. Used by the writer to + * detect whether an append whose response was lost actually committed. + */ + async payloadAt(threadId: string, seq: number): Promise { + const row = await this.findOne({ where: { threadId, seq } }); + return row?.payload ?? null; + } + + async getAfter(threadId: string, afterSeq: number): Promise { + const rows = await this.find({ + where: { threadId, seq: MoreThan(afterSeq) }, + order: { seq: 'ASC' }, + }); + return rows.map((r) => ({ id: r.seq, event: jsonParse(r.payload) })); + } + + async getForRuns(threadId: string, runIds: string[]): Promise { + if (runIds.length === 0) return []; + const rows = await this.createQueryBuilder('e') + .where('e.threadId = :threadId', { threadId }) + .andWhere('e.runId IN (:...runIds)', { runIds }) + .orderBy('e.seq', 'ASC') + .getMany(); + return rows.map((r) => jsonParse(r.payload)); + } +}