mirror of
https://github.com/n8n-io/n8n.git
synced 2026-09-01 05:38:33 +08:00
feat(core): Add the Instance AI durable event-log store and writer (no-changelog) (#33910)
Co-authored-by: Danny Martini <danny@n8n.io> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Raúl Gómez Morales
committed by
GitHub
GitHub
parent
9c3f7fa44b
commit
05b412cd28
@@ -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
|
||||
|
||||
@@ -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)
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
# instance_ai_events
|
||||
|
||||
## Description
|
||||
|
||||
<details>
|
||||
<summary><strong>Table Definition</strong></summary>
|
||||
|
||||
```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"))
|
||||
```
|
||||
|
||||
</details>
|
||||
|
||||
## 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)
|
||||
@@ -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
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
];
|
||||
|
||||
@@ -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 };
|
||||
|
||||
@@ -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';
|
||||
|
||||
@@ -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 }) => {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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<void> | undefined;
|
||||
|
||||
async maxSeq(_threadId: string): Promise<number> {
|
||||
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<string | null> {
|
||||
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>();
|
||||
logger.scoped.mockReturnValue(logger);
|
||||
const metrics = new DurableLogMetrics(mock<EventService>());
|
||||
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<DrainedEvent[]> {
|
||||
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.',
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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<string, PendingEntry[]>();
|
||||
|
||||
/** In-flight drain per thread, awaited by flush(). */
|
||||
private readonly draining = new Map<string, Promise<void>>();
|
||||
|
||||
/** Last assigned seq per thread; lazily seeded from MAX(seq) in the DB. */
|
||||
private readonly lastSeq = new Map<string, number>();
|
||||
|
||||
/** Open text/reasoning blocks per thread, keyed `${runId}:${agentId}`. */
|
||||
private readonly buffers = new Map<string, Map<string, CoalesceBuffer>>();
|
||||
|
||||
private readonly emitters = new Map<string, EmitFn>();
|
||||
|
||||
/**
|
||||
* 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<string, object>();
|
||||
|
||||
/** Per-thread idle timers driving the trailing-delta flush. */
|
||||
private readonly idleFlushTimers = new Map<string, NodeJS.Timeout>();
|
||||
|
||||
/** 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<StoredEvent[]> {
|
||||
return await this.repo.getAfter(threadId, afterSeq);
|
||||
}
|
||||
|
||||
async getEventsForRuns(threadId: string, runIds: string[]): Promise<InstanceAiEvent[]> {
|
||||
return await this.repo.getForRuns(threadId, runIds);
|
||||
}
|
||||
|
||||
async getNextEventId(threadId: string): Promise<number> {
|
||||
// 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<void> {
|
||||
const hasBuffers = (this.buffers.get(threadId)?.size ?? 0) > 0;
|
||||
if (!this.pendingByThread.has(threadId) && !this.draining.has(threadId) && !hasBuffers) {
|
||||
return;
|
||||
}
|
||||
await new Promise<void>((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<void> {
|
||||
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<void> {
|
||||
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<number | undefined> {
|
||||
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<boolean> {
|
||||
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<number> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -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 });
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
];
|
||||
}
|
||||
|
||||
|
||||
@@ -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';
|
||||
|
||||
@@ -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<InstanceAiEventLogEntry> {
|
||||
constructor(dataSource: DataSource) {
|
||||
super(InstanceAiEventLogEntry, dataSource.manager);
|
||||
}
|
||||
|
||||
/** Highest seq assigned for a thread, 0 when the log is empty. */
|
||||
async maxSeq(threadId: string): Promise<number> {
|
||||
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<number> {
|
||||
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<string | null> {
|
||||
const row = await this.findOne({ where: { threadId, seq } });
|
||||
return row?.payload ?? null;
|
||||
}
|
||||
|
||||
async getAfter(threadId: string, afterSeq: number): Promise<StoredEvent[]> {
|
||||
const rows = await this.find({
|
||||
where: { threadId, seq: MoreThan(afterSeq) },
|
||||
order: { seq: 'ASC' },
|
||||
});
|
||||
return rows.map((r) => ({ id: r.seq, event: jsonParse<InstanceAiEvent>(r.payload) }));
|
||||
}
|
||||
|
||||
async getForRuns(threadId: string, runIds: string[]): Promise<InstanceAiEvent[]> {
|
||||
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<InstanceAiEvent>(r.payload));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user