mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 13:39:44 +02:00
fix(core): Include sub-workflow tool calls in the workflow dependency index (#38062)
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
c09e00884c
commit
d3c8eee07e
|
|
@ -70,7 +70,10 @@ import { TestRun } from './test-run.ee';
|
|||
import { User } from './user';
|
||||
import { Variables } from './variables';
|
||||
import { WebhookEntity } from './webhook-entity';
|
||||
import { WorkflowDependency } from './workflow-dependency-entity';
|
||||
import {
|
||||
WorkflowDependency,
|
||||
WORKFLOW_DEPENDENCY_INDEX_VERSION,
|
||||
} from './workflow-dependency-entity';
|
||||
import { WorkflowEntity } from './workflow-entity';
|
||||
import { WorkflowHistory } from './workflow-history';
|
||||
import {
|
||||
|
|
@ -160,6 +163,7 @@ export {
|
|||
TagEntity,
|
||||
User,
|
||||
WorkflowDependency,
|
||||
WORKFLOW_DEPENDENCY_INDEX_VERSION,
|
||||
WorkflowEntity,
|
||||
WorkflowStatistics,
|
||||
WorkflowTagMapping,
|
||||
|
|
|
|||
|
|
@ -11,6 +11,13 @@ import {
|
|||
import { WithCreatedAt } from './abstract-entity';
|
||||
import type { WorkflowEntity } from './workflow-entity';
|
||||
|
||||
// Version of the index structure. When a change to the indexer alters what it
|
||||
// extracts, bump this version. The staleness scan behind the rebuild on server
|
||||
// start then reindexes every workflow whose rows carry an older version.
|
||||
// Version 2: the indexer records `workflowCall` rows for sub-workflow tool and
|
||||
// retriever nodes, not only for the Execute Sub-workflow node.
|
||||
export const WORKFLOW_DEPENDENCY_INDEX_VERSION = 2;
|
||||
|
||||
export type DependencyType =
|
||||
| 'credentialId'
|
||||
| 'dataTableId'
|
||||
|
|
|
|||
|
|
@ -1,12 +1,18 @@
|
|||
import { DatabaseConfig } from '@n8n/config';
|
||||
import { Service } from '@n8n/di';
|
||||
import { DataSource, EntityManager, IsNull, LessThan, Repository, Not } from '@n8n/typeorm';
|
||||
import {
|
||||
DataSource,
|
||||
EntityManager,
|
||||
IsNull,
|
||||
LessThan,
|
||||
LessThanOrEqual,
|
||||
Repository,
|
||||
Not,
|
||||
} from '@n8n/typeorm';
|
||||
|
||||
import { SharedWorkflowRepository } from './shared-workflow.repository';
|
||||
import type { User } from '../entities';
|
||||
import { WorkflowDependency, WorkflowEntity } from '../entities';
|
||||
|
||||
const INDEX_VERSION_ID = 1;
|
||||
import { WorkflowDependency, WorkflowEntity, WORKFLOW_DEPENDENCY_INDEX_VERSION } from '../entities';
|
||||
|
||||
/**
|
||||
* Which workflows an aggregate over the index may span. The roles come from the caller
|
||||
|
|
@ -45,7 +51,7 @@ export class WorkflowDependencies {
|
|||
workflowId: this.workflowId,
|
||||
workflowVersionId: this.workflowVersionId,
|
||||
publishedVersionId: this.publishedVersionId,
|
||||
indexVersionId: INDEX_VERSION_ID,
|
||||
indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION,
|
||||
});
|
||||
this.dependencies.push(dep);
|
||||
}
|
||||
|
|
@ -203,17 +209,31 @@ export class WorkflowDependencyRepository extends Repository<WorkflowDependency>
|
|||
dependencies: WorkflowDependencies,
|
||||
tx: EntityManager,
|
||||
): Promise<boolean> {
|
||||
const deleteResult = await tx.delete(WorkflowDependency, {
|
||||
// NOTE: this relies on the fact that we only want to track the latest published version or draft dependencies.
|
||||
// If we're updating published dependencies, checking for Not Null works because we don't actually
|
||||
// care about the specific previous published version id.
|
||||
const publishedVersionCondition = dependencies.publishedVersionId ? Not(IsNull()) : IsNull();
|
||||
|
||||
// Replace rows from an older workflow version, and rows an older indexer
|
||||
// version wrote for the current workflow version.
|
||||
const outdatedVersionResult = await tx.delete(WorkflowDependency, {
|
||||
workflowId,
|
||||
workflowVersionId: LessThan(dependencies.workflowVersionId),
|
||||
// NOTE: this relies on the fact that we only want to track the latest published version or draft dependencies.
|
||||
// If we're updating published dependencies, checking for Not Null works because we don't actually
|
||||
// care about the specific previous published version id.
|
||||
publishedVersionId: dependencies.publishedVersionId ? Not(IsNull()) : IsNull(),
|
||||
publishedVersionId: publishedVersionCondition,
|
||||
});
|
||||
// The workflow version bound stops a stale request from replacing rows a
|
||||
// newer workflow version wrote before this indexer version deployed.
|
||||
const outdatedIndexResult = await tx.delete(WorkflowDependency, {
|
||||
workflowId,
|
||||
workflowVersionId: LessThanOrEqual(dependencies.workflowVersionId),
|
||||
indexVersionId: LessThan(WORKFLOW_DEPENDENCY_INDEX_VERSION),
|
||||
publishedVersionId: publishedVersionCondition,
|
||||
});
|
||||
const deletedCount =
|
||||
(outdatedVersionResult.affected ?? 0) + (outdatedIndexResult.affected ?? 0);
|
||||
|
||||
// If we deleted something, the incoming version is newer - proceed with insert
|
||||
if (deleteResult.affected && deleteResult.affected > 0) {
|
||||
// If we deleted something, the incoming rows are newer - proceed with insert
|
||||
if (deletedCount > 0) {
|
||||
// NOTE: we cast to any[] because TypeORM doesn't like the JSON column.
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
await tx.insert(WorkflowDependency, dependencies.dependencies as any[]);
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ import {
|
|||
WorkflowEntity,
|
||||
WorkflowTagMapping,
|
||||
WorkflowDependency,
|
||||
WORKFLOW_DEPENDENCY_INDEX_VERSION,
|
||||
User,
|
||||
} from '../entities';
|
||||
import { SharedWorkflow } from '../entities/shared-workflow';
|
||||
|
|
@ -1750,8 +1751,9 @@ export class WorkflowRepository extends BaseRepository<WorkflowEntity> {
|
|||
}
|
||||
|
||||
/**
|
||||
* Find workflows that need draft indexing - either unindexed (no draft entries in workflow_dependency)
|
||||
* or outdated (versionCounter > workflowVersionId in workflow_dependency for drafts).
|
||||
* Find workflows that need draft indexing - unindexed (no draft entries in workflow_dependency),
|
||||
* outdated (versionCounter > workflowVersionId in workflow_dependency for drafts), or indexed
|
||||
* by an older indexer version (indexVersionId < WORKFLOW_DEPENDENCY_INDEX_VERSION).
|
||||
*
|
||||
* NOTE: we use a simple batch limit instead of proper pagination because we use this
|
||||
* method to retrieve workflows and then index them immediately - so they won't be returned
|
||||
|
|
@ -1762,6 +1764,7 @@ export class WorkflowRepository extends BaseRepository<WorkflowEntity> {
|
|||
const qb = this.createQueryBuilder('workflow');
|
||||
const workflowIdAlias = 'workflowId';
|
||||
const maxVersionIdAlias = 'maxVersionId';
|
||||
const minIndexVersionAlias = 'minIndexVersion';
|
||||
const depAlias = 'dep';
|
||||
|
||||
// Only select columns needed for indexing to avoid loading large unused
|
||||
|
|
@ -1779,6 +1782,7 @@ export class WorkflowRepository extends BaseRepository<WorkflowEntity> {
|
|||
subQuery
|
||||
.select('wd.workflowId', workflowIdAlias)
|
||||
.addSelect('MAX(wd.workflowVersionId)', maxVersionIdAlias)
|
||||
.addSelect('MIN(wd.indexVersionId)', minIndexVersionAlias)
|
||||
.from(WorkflowDependency, 'wd')
|
||||
// Only consider draft dependencies (publishedVersionId IS NULL)
|
||||
.where('wd.publishedVersionId IS NULL')
|
||||
|
|
@ -1792,9 +1796,12 @@ export class WorkflowRepository extends BaseRepository<WorkflowEntity> {
|
|||
// Include workflows that are either:
|
||||
// 1. Unindexed (no draft dependency entries exist)
|
||||
// 2. Outdated (workflow version is newer than indexed version)
|
||||
qb.where(`${qb.escape(depAlias)}.${qb.escape(workflowIdAlias)} IS NULL`).orWhere(
|
||||
`workflow.versionCounter > ${qb.escape(depAlias)}.${qb.escape(maxVersionIdAlias)}`,
|
||||
);
|
||||
// 3. Indexed by an older indexer version
|
||||
qb.where(`${qb.escape(depAlias)}.${qb.escape(workflowIdAlias)} IS NULL`)
|
||||
.orWhere(`workflow.versionCounter > ${qb.escape(depAlias)}.${qb.escape(maxVersionIdAlias)}`)
|
||||
.orWhere(`${qb.escape(depAlias)}.${qb.escape(minIndexVersionAlias)} < :indexVersion`, {
|
||||
indexVersion: WORKFLOW_DEPENDENCY_INDEX_VERSION,
|
||||
});
|
||||
if (batchSize) {
|
||||
qb.limit(batchSize);
|
||||
}
|
||||
|
|
@ -1806,7 +1813,8 @@ export class WorkflowRepository extends BaseRepository<WorkflowEntity> {
|
|||
* Find active workflows that need published version indexing.
|
||||
* These are workflows where:
|
||||
* - activeVersionId IS NOT NULL (workflow is active/published)
|
||||
* - No dependency rows exist with matching publishedVersionId = activeVersionId
|
||||
* - No dependency rows exist with matching publishedVersionId = activeVersionId,
|
||||
* or the matching rows were written by an older indexer version
|
||||
*
|
||||
* This includes the activeVersion relation for efficiency.
|
||||
*/
|
||||
|
|
@ -1816,6 +1824,7 @@ export class WorkflowRepository extends BaseRepository<WorkflowEntity> {
|
|||
const qb = this.createQueryBuilder('workflow');
|
||||
const depAlias = 'dep';
|
||||
const publishedVersionIdAlias = 'publishedVersionId';
|
||||
const minIndexVersionAlias = 'minIndexVersion';
|
||||
|
||||
// Only select columns needed for indexing to avoid loading large unused
|
||||
// JSON columns (connections, staticData, pinData) that can cause OOM.
|
||||
|
|
@ -1829,6 +1838,7 @@ export class WorkflowRepository extends BaseRepository<WorkflowEntity> {
|
|||
return subQuery
|
||||
.select('wd.workflowId', 'workflowId')
|
||||
.addSelect('wd.publishedVersionId', publishedVersionIdAlias)
|
||||
.addSelect('MIN(wd.indexVersionId)', minIndexVersionAlias)
|
||||
.from(WorkflowDependency, 'wd')
|
||||
.where('wd.publishedVersionId IS NOT NULL')
|
||||
.groupBy('wd.workflowId')
|
||||
|
|
@ -1838,9 +1848,11 @@ export class WorkflowRepository extends BaseRepository<WorkflowEntity> {
|
|||
`workflow.id = ${qb.escape(depAlias)}.${qb.escape('workflowId')} AND workflow.activeVersionId = ${qb.escape(depAlias)}.${qb.escape(publishedVersionIdAlias)}`,
|
||||
);
|
||||
|
||||
// Only include active workflows with no matching published version dependency
|
||||
// Only include active workflows whose published version dependency is
|
||||
// missing or was written by an older indexer version
|
||||
qb.where('workflow.activeVersionId IS NOT NULL').andWhere(
|
||||
`${qb.escape(depAlias)}.${qb.escape(publishedVersionIdAlias)} IS NULL`,
|
||||
`(${qb.escape(depAlias)}.${qb.escape(publishedVersionIdAlias)} IS NULL OR ${qb.escape(depAlias)}.${qb.escape(minIndexVersionAlias)} < :indexVersion)`,
|
||||
{ indexVersion: WORKFLOW_DEPENDENCY_INDEX_VERSION },
|
||||
);
|
||||
|
||||
// Include the published version's nodes for indexing (skip connections to save memory).
|
||||
|
|
|
|||
|
|
@ -239,6 +239,114 @@ describe('WorkflowIndexService', () => {
|
|||
);
|
||||
});
|
||||
|
||||
it('should create workflowCall dependencies for sub-workflow tool and retriever nodes', async () => {
|
||||
mockWorkflowDependencyRepository.updateDependenciesForWorkflow.mockResolvedValue(true);
|
||||
|
||||
const workflow = createWorkflow([
|
||||
createNode({
|
||||
id: 'node-1',
|
||||
type: '@n8n/n8n-nodes-langchain.toolWorkflow',
|
||||
typeVersion: 2.2,
|
||||
parameters: { source: 'database', workflowId: { mode: 'list', value: 'sub-workflow-1' } },
|
||||
}),
|
||||
// toolWorkflow <= 1.1 stores the workflowId as a plain string.
|
||||
createNode({
|
||||
id: 'node-2',
|
||||
type: '@n8n/n8n-nodes-langchain.toolWorkflow',
|
||||
typeVersion: 1,
|
||||
parameters: { source: 'database', workflowId: 'sub-workflow-2' },
|
||||
}),
|
||||
createNode({
|
||||
id: 'node-3',
|
||||
type: '@n8n/n8n-nodes-langchain.retrieverWorkflow',
|
||||
parameters: { source: 'database', workflowId: 'sub-workflow-3' },
|
||||
}),
|
||||
]);
|
||||
|
||||
await service.updateIndexForDraft(workflow);
|
||||
|
||||
expect(mockWorkflowDependencyRepository.updateDependenciesForWorkflow).toHaveBeenCalledWith(
|
||||
'workflow-123',
|
||||
expect.objectContaining({
|
||||
dependencies: expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
dependencyType: 'workflowCall',
|
||||
dependencyKey: 'sub-workflow-1',
|
||||
dependencyInfo: { nodeId: 'node-1', nodeVersion: 2.2 },
|
||||
}),
|
||||
expect.objectContaining({
|
||||
dependencyType: 'workflowCall',
|
||||
dependencyKey: 'sub-workflow-2',
|
||||
dependencyInfo: { nodeId: 'node-2', nodeVersion: 1 },
|
||||
}),
|
||||
expect.objectContaining({
|
||||
dependencyType: 'workflowCall',
|
||||
dependencyKey: 'sub-workflow-3',
|
||||
dependencyInfo: { nodeId: 'node-3', nodeVersion: 1 },
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('should not create workflowCall dependencies for sub-workflow tool nodes with an inline workflow', async () => {
|
||||
mockWorkflowDependencyRepository.updateDependenciesForWorkflow.mockResolvedValue(true);
|
||||
|
||||
const workflow = createWorkflow([
|
||||
createNode({
|
||||
id: 'node-1',
|
||||
type: '@n8n/n8n-nodes-langchain.toolWorkflow',
|
||||
parameters: { source: 'parameter', workflowJson: '{}' },
|
||||
}),
|
||||
]);
|
||||
|
||||
await service.updateIndexForDraft(workflow);
|
||||
|
||||
expect(mockWorkflowDependencyRepository.updateDependenciesForWorkflow).toHaveBeenCalledWith(
|
||||
'workflow-123',
|
||||
expect.objectContaining({
|
||||
dependencies: expect.not.arrayContaining([
|
||||
expect.objectContaining({
|
||||
dependencyType: 'workflowCall',
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('should not create workflowCall dependencies for expression-based workflow IDs', async () => {
|
||||
mockWorkflowDependencyRepository.updateDependenciesForWorkflow.mockResolvedValue(true);
|
||||
|
||||
const workflow = createWorkflow([
|
||||
createNode({
|
||||
id: 'node-1',
|
||||
type: 'n8n-nodes-base.executeWorkflow',
|
||||
parameters: { workflowId: '={{ $json.workflowId }}' },
|
||||
}),
|
||||
createNode({
|
||||
id: 'node-2',
|
||||
type: '@n8n/n8n-nodes-langchain.toolWorkflow',
|
||||
parameters: {
|
||||
source: 'database',
|
||||
workflowId: { mode: 'id', value: '={{ $json.workflowId }}' },
|
||||
},
|
||||
}),
|
||||
]);
|
||||
|
||||
await service.updateIndexForDraft(workflow);
|
||||
|
||||
expect(mockWorkflowDependencyRepository.updateDependenciesForWorkflow).toHaveBeenCalledWith(
|
||||
'workflow-123',
|
||||
expect.objectContaining({
|
||||
dependencies: expect.not.arrayContaining([
|
||||
expect.objectContaining({
|
||||
dependencyType: 'workflowCall',
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('should handle multiple credentials on a single node', async () => {
|
||||
mockWorkflowDependencyRepository.updateDependenciesForWorkflow.mockResolvedValue(true);
|
||||
|
||||
|
|
|
|||
|
|
@ -5,7 +5,13 @@ import { WorkflowDependencies, WorkflowDependencyRepository, WorkflowRepository
|
|||
import { Service } from '@n8n/di';
|
||||
import { ErrorReporter, SpanStatus, Tracing } from 'n8n-core';
|
||||
import { ensureError } from '@n8n/utils/errors/ensure-error';
|
||||
import { DATA_TABLE_NODE_TYPES, INode, IWorkflowBase, IWorkflowSettings } from 'n8n-workflow';
|
||||
import {
|
||||
DATA_TABLE_NODE_TYPES,
|
||||
INode,
|
||||
IWorkflowBase,
|
||||
IWorkflowSettings,
|
||||
isNodeWithWorkflowSelector,
|
||||
} from 'n8n-workflow';
|
||||
|
||||
import { EventService } from '@/events/event.service';
|
||||
|
||||
|
|
@ -307,13 +313,18 @@ export class WorkflowIndexService {
|
|||
}
|
||||
|
||||
private addWorkflowCallDependencies(node: INode, dependencyUpdates: WorkflowDependencies): void {
|
||||
if (node.type !== 'n8n-nodes-base.executeWorkflow') {
|
||||
// Covers the Execute Sub-workflow node, the sub-workflow tool, and the workflow retriever.
|
||||
if (!isNodeWithWorkflowSelector(node)) {
|
||||
return;
|
||||
}
|
||||
const calledWorkflowId: string | undefined = this.getCalledWorkflowIdFrom(node);
|
||||
if (!calledWorkflowId) {
|
||||
return;
|
||||
}
|
||||
// Expressions resolve at runtime, so they are not static dependencies.
|
||||
if (calledWorkflowId.trim().startsWith('=')) {
|
||||
return;
|
||||
}
|
||||
dependencyUpdates.add({
|
||||
dependencyType: 'workflowCall',
|
||||
dependencyKey: calledWorkflowId,
|
||||
|
|
@ -378,7 +389,7 @@ export class WorkflowIndexService {
|
|||
return node.parameters['workflowId']['value'];
|
||||
}
|
||||
this.errorReporter.warn(
|
||||
`While indexing, could not determine called workflow ID from executeWorkflow node ${node.id}`,
|
||||
`While indexing, could not determine called workflow ID from ${node.type} node ${node.id}`,
|
||||
{ extra: node.parameters },
|
||||
);
|
||||
return undefined;
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ import {
|
|||
WorkflowDependencyRepository,
|
||||
WorkflowDependencies,
|
||||
WorkflowRepository,
|
||||
WORKFLOW_DEPENDENCY_INDEX_VERSION,
|
||||
} from '@n8n/db';
|
||||
import { Container } from '@n8n/di';
|
||||
|
||||
|
|
@ -73,7 +74,7 @@ describe('WorkflowDependencyRepository', () => {
|
|||
dependencyType: 'credentialId',
|
||||
dependencyKey: 'cred-123',
|
||||
dependencyInfo: { name: 'Test Credential' },
|
||||
indexVersionId: 1,
|
||||
indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION,
|
||||
});
|
||||
expect(savedDependencies[1]).toMatchObject({
|
||||
workflowId: workflow.id,
|
||||
|
|
@ -81,7 +82,7 @@ describe('WorkflowDependencyRepository', () => {
|
|||
dependencyType: 'nodeType',
|
||||
dependencyKey: 'n8n-nodes-base.httpRequest',
|
||||
dependencyInfo: null,
|
||||
indexVersionId: 1,
|
||||
indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION,
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -179,6 +180,100 @@ describe('WorkflowDependencyRepository', () => {
|
|||
expect(savedDependencies[0].workflowVersionId).toBe(2);
|
||||
});
|
||||
|
||||
it('should replace rows an older indexer version wrote for the same workflow version', async () => {
|
||||
//
|
||||
// ARRANGE
|
||||
//
|
||||
const workflow = await createWorkflow({ versionId: 'v1', nodes: [] });
|
||||
|
||||
// Seed rows directly, as an older indexer version would have written them.
|
||||
await workflowDependencyRepository.insert({
|
||||
workflowId: workflow.id,
|
||||
workflowVersionId: 1,
|
||||
publishedVersionId: null,
|
||||
dependencyType: 'credentialId',
|
||||
dependencyKey: 'cred-old',
|
||||
dependencyInfo: null,
|
||||
indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION - 1,
|
||||
});
|
||||
|
||||
const sameVersionDeps = new WorkflowDependencies(workflow.id, 1);
|
||||
sameVersionDeps.add({
|
||||
dependencyType: 'credentialId',
|
||||
dependencyKey: 'cred-new',
|
||||
dependencyInfo: null,
|
||||
});
|
||||
|
||||
//
|
||||
// ACT
|
||||
//
|
||||
const result = await workflowDependencyRepository.updateDependenciesForWorkflow(
|
||||
workflow.id,
|
||||
sameVersionDeps,
|
||||
);
|
||||
|
||||
//
|
||||
// ASSERT
|
||||
//
|
||||
expect(result).toBe(true);
|
||||
const savedDependencies = await workflowDependencyRepository.find({
|
||||
where: { workflowId: workflow.id },
|
||||
});
|
||||
expect(savedDependencies).toHaveLength(1);
|
||||
expect(savedDependencies[0]).toMatchObject({
|
||||
dependencyKey: 'cred-new',
|
||||
workflowVersionId: 1,
|
||||
indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION,
|
||||
});
|
||||
});
|
||||
|
||||
it('should not let a stale request replace rows a newer workflow version wrote', async () => {
|
||||
//
|
||||
// ARRANGE
|
||||
//
|
||||
const workflow = await createWorkflow({ versionId: 'v1', nodes: [] });
|
||||
|
||||
// Rows an older indexer version wrote for a newer workflow version.
|
||||
await workflowDependencyRepository.insert({
|
||||
workflowId: workflow.id,
|
||||
workflowVersionId: 5,
|
||||
publishedVersionId: null,
|
||||
dependencyType: 'credentialId',
|
||||
dependencyKey: 'cred-newer',
|
||||
dependencyInfo: null,
|
||||
indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION - 1,
|
||||
});
|
||||
|
||||
const staleDeps = new WorkflowDependencies(workflow.id, 3);
|
||||
staleDeps.add({
|
||||
dependencyType: 'credentialId',
|
||||
dependencyKey: 'cred-stale',
|
||||
dependencyInfo: null,
|
||||
});
|
||||
|
||||
//
|
||||
// ACT
|
||||
//
|
||||
const result = await workflowDependencyRepository.updateDependenciesForWorkflow(
|
||||
workflow.id,
|
||||
staleDeps,
|
||||
);
|
||||
|
||||
//
|
||||
// ASSERT
|
||||
//
|
||||
expect(result).toBe(false);
|
||||
const savedDependencies = await workflowDependencyRepository.find({
|
||||
where: { workflowId: workflow.id },
|
||||
});
|
||||
expect(savedDependencies).toHaveLength(1);
|
||||
expect(savedDependencies[0]).toMatchObject({
|
||||
dependencyKey: 'cred-newer',
|
||||
workflowVersionId: 5,
|
||||
indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION - 1,
|
||||
});
|
||||
});
|
||||
|
||||
it('should prevent races between concurrent updates', async () => {
|
||||
//
|
||||
// ARRANGE
|
||||
|
|
@ -594,7 +689,7 @@ describe('WorkflowDependencyRepository node usage', () => {
|
|||
dependencyType: 'nodeType' as const,
|
||||
dependencyKey: index < anthropicCount ? ANTHROPIC : OPENAI,
|
||||
dependencyInfo: null,
|
||||
indexVersionId: 1,
|
||||
indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION,
|
||||
})),
|
||||
async (chunk) => await repository.insert(chunk),
|
||||
);
|
||||
|
|
|
|||
|
|
@ -8,7 +8,12 @@ import {
|
|||
getWorkflowById,
|
||||
setActiveVersion,
|
||||
} from '@n8n/backend-test-utils';
|
||||
import { WorkflowRepository, WorkflowDependencyRepository, WorkflowDependencies } from '@n8n/db';
|
||||
import {
|
||||
WorkflowRepository,
|
||||
WorkflowDependencyRepository,
|
||||
WorkflowDependencies,
|
||||
WORKFLOW_DEPENDENCY_INDEX_VERSION,
|
||||
} from '@n8n/db';
|
||||
import { Container } from '@n8n/di';
|
||||
import type { Scope } from '@n8n/permissions';
|
||||
|
||||
|
|
@ -484,6 +489,51 @@ describe('WorkflowRepository', () => {
|
|||
expect(workflowIds).not.toContain(workflow3.id);
|
||||
});
|
||||
|
||||
it('should return workflows whose rows an older indexer version wrote', async () => {
|
||||
//
|
||||
// ARRANGE
|
||||
//
|
||||
const workflowRepository = Container.get(WorkflowRepository);
|
||||
const workflowDependencyRepository = Container.get(WorkflowDependencyRepository);
|
||||
|
||||
// Indexed at the current version counter, but by an older indexer version.
|
||||
const outdatedWorkflow = await createWorkflow({ versionCounter: 3, nodes: [] });
|
||||
await workflowDependencyRepository.insert({
|
||||
workflowId: outdatedWorkflow.id,
|
||||
workflowVersionId: 3,
|
||||
publishedVersionId: null,
|
||||
dependencyType: 'nodeType',
|
||||
dependencyKey: 'n8n-nodes-base.httpRequest',
|
||||
dependencyInfo: null,
|
||||
indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION - 1,
|
||||
});
|
||||
|
||||
// Indexed by the current indexer version.
|
||||
const currentWorkflow = await createWorkflow({ versionCounter: 3, nodes: [] });
|
||||
const dependencies = new WorkflowDependencies(currentWorkflow.id, 3);
|
||||
dependencies.add({
|
||||
dependencyType: 'nodeType',
|
||||
dependencyKey: 'n8n-nodes-base.httpRequest',
|
||||
dependencyInfo: null,
|
||||
});
|
||||
await workflowDependencyRepository.updateDependenciesForWorkflow(
|
||||
currentWorkflow.id,
|
||||
dependencies,
|
||||
);
|
||||
|
||||
//
|
||||
// ACT
|
||||
//
|
||||
const workflowsNeedingIndexing = await workflowRepository.findWorkflowsNeedingIndexing();
|
||||
|
||||
//
|
||||
// ASSERT
|
||||
//
|
||||
const workflowIds = workflowsNeedingIndexing.map((w) => w.id);
|
||||
expect(workflowIds).toContain(outdatedWorkflow.id);
|
||||
expect(workflowIds).not.toContain(currentWorkflow.id);
|
||||
});
|
||||
|
||||
it('should respect the batch size limit', async () => {
|
||||
//
|
||||
// ARRANGE
|
||||
|
|
@ -509,6 +559,62 @@ describe('WorkflowRepository', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('findWorkflowsNeedingPublishedVersionIndexing', () => {
|
||||
it('should return active workflows with missing or outdated published rows', async () => {
|
||||
//
|
||||
// ARRANGE
|
||||
//
|
||||
const workflowRepository = Container.get(WorkflowRepository);
|
||||
const workflowDependencyRepository = Container.get(WorkflowDependencyRepository);
|
||||
|
||||
// Active, but no published dependency rows exist.
|
||||
const unindexedWorkflow = await createActiveWorkflow();
|
||||
|
||||
// Published rows exist, but an older indexer version wrote them.
|
||||
const outdatedWorkflow = await createActiveWorkflow();
|
||||
await workflowDependencyRepository.insert({
|
||||
workflowId: outdatedWorkflow.id,
|
||||
workflowVersionId: 1,
|
||||
publishedVersionId: outdatedWorkflow.activeVersionId,
|
||||
dependencyType: 'nodeType',
|
||||
dependencyKey: 'n8n-nodes-base.httpRequest',
|
||||
dependencyInfo: null,
|
||||
indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION - 1,
|
||||
});
|
||||
|
||||
// Published rows exist at the current indexer version.
|
||||
const currentWorkflow = await createActiveWorkflow();
|
||||
const dependencies = new WorkflowDependencies(
|
||||
currentWorkflow.id,
|
||||
1,
|
||||
currentWorkflow.activeVersionId,
|
||||
);
|
||||
dependencies.add({
|
||||
dependencyType: 'nodeType',
|
||||
dependencyKey: 'n8n-nodes-base.httpRequest',
|
||||
dependencyInfo: null,
|
||||
});
|
||||
await workflowDependencyRepository.updateDependenciesForWorkflow(
|
||||
currentWorkflow.id,
|
||||
dependencies,
|
||||
);
|
||||
|
||||
//
|
||||
// ACT
|
||||
//
|
||||
const workflowsNeedingIndexing =
|
||||
await workflowRepository.findWorkflowsNeedingPublishedVersionIndexing();
|
||||
|
||||
//
|
||||
// ASSERT
|
||||
//
|
||||
const workflowIds = workflowsNeedingIndexing.map((w) => w.id);
|
||||
expect(workflowIds).toContain(unindexedWorkflow.id);
|
||||
expect(workflowIds).toContain(outdatedWorkflow.id);
|
||||
expect(workflowIds).not.toContain(currentWorkflow.id);
|
||||
});
|
||||
});
|
||||
|
||||
describe('findWorkflowsWithNodeType', () => {
|
||||
it('should return workflows that have the specified node type in workflow_dependency', async () => {
|
||||
//
|
||||
|
|
|
|||
|
|
@ -208,7 +208,7 @@ describe('WorkflowIndexService Integration', () => {
|
|||
nodeId: 'node-1',
|
||||
nodeVersion: 1,
|
||||
},
|
||||
indexVersionId: 1,
|
||||
indexVersionId: 2,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user