opensource/n8n
3
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-09-24 23:22:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(MongoDBVectorStore Node): Replace MongoDB Atlas Vector Store singleton with per-execution client (#26906)

Browse Source
This commit is contained in:
Albert Alises
2026-03-11 15:35:12 +00:00
committed by GitHub
parent f6c4f28e4a
commit 33391e6e15
2 changed files with 123 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/@n8n/nodes-langchain/nodes/vector_store/VectorStoreMongoDBAtlas/VectorStoreMongoDBAtlas.node.test.ts
+93 -144
View File
@@ -5,12 +5,12 @@ import type { ILoadOptionsFunctions, ISupplyDataFunctions } from 'n8n-workflow';
import {
EMBEDDING_NAME,
getCollectionName,
getCollections,
getEmbeddingFieldName,
getFilterValue,
getMetadataFieldName,
getMongoClient,
createMongoClient,
getVectorIndexName,
mongoConfig,
METADATA_FIELD_NAME,
MONGODB_COLLECTION_NAME,
VECTOR_INDEX_NAME,
@@ -30,34 +30,47 @@ describe('VectorStoreMongoDBAtlas', () => {
jest.resetAllMocks();
});
describe('.getMongoClient', () => {
describe('.createMongoClient', () => {
const mockContext = mock<ISupplyDataFunctions>({
getCredentials: jest.fn(),
});
const mockClient1 = {
connect: jest.fn().mockResolvedValue(undefined),
close: jest.fn().mockResolvedValue(undefined),
};
const mockClient2 = {
const mockClient = {
connect: jest.fn().mockResolvedValue(undefined),
close: jest.fn().mockResolvedValue(undefined),
};
const MockMongoClient = MongoClient as jest.MockedClass<typeof MongoClient>;
beforeEach(() => {
mongoConfig.client = null;
mongoConfig.connectionString = '';
});
it('should reuse the same client when connection string is unchanged', async () => {
MockMongoClient.mockImplementation(() => mockClient1 as unknown as MongoClient);
it('should create a fresh client on every call', async () => {
const mockClient2 = {
connect: jest.fn().mockResolvedValue(undefined),
close: jest.fn().mockResolvedValue(undefined),
};
MockMongoClient.mockImplementationOnce(
() => mockClient as unknown as MongoClient,
).mockImplementationOnce(() => mockClient2 as unknown as MongoClient);
mockContext.getCredentials.mockResolvedValue({
configurationType: 'connectionString',
connectionString: 'mongodb://localhost:27017',
});
const client1 = await getMongoClient(mockContext, 1.1);
const client2 = await getMongoClient(mockContext, 1.1);
const client1 = await createMongoClient(mockContext, 1.1);
const client2 = await createMongoClient(mockContext, 1.1);
expect(MockMongoClient).toHaveBeenCalledTimes(2);
expect(mockClient.connect).toHaveBeenCalledTimes(1);
expect(mockClient2.connect).toHaveBeenCalledTimes(1);
expect(client1).toBe(mockClient);
expect(client2).toBe(mockClient2);
});
it('should create client with connectionString config', async () => {
MockMongoClient.mockImplementation(() => mockClient as unknown as MongoClient);
mockContext.getCredentials.mockResolvedValue({
configurationType: 'connectionString',
connectionString: 'mongodb://localhost:27017',
});
const client = await createMongoClient(mockContext, 1.1);
expect(MockMongoClient).toHaveBeenCalledTimes(1);
expect(MockMongoClient).toHaveBeenCalledWith('mongodb://localhost:27017', {
@@ -67,55 +80,12 @@ describe('VectorStoreMongoDBAtlas', () => {
version: '1.1',
},
});
expect(mockClient1.connect).toHaveBeenCalledTimes(1);
expect(mockClient1.close).not.toHaveBeenCalled();
expect(mockClient2.connect).not.toHaveBeenCalled();
expect(client1).toBe(mockClient1);
expect(client2).toBe(mockClient1);
});
it('should create new client when connection string changes', async () => {
MockMongoClient.mockImplementationOnce(
() => mockClient1 as unknown as MongoClient,
).mockImplementationOnce(() => mockClient2 as unknown as MongoClient);
mockContext.getCredentials
.mockResolvedValueOnce({
configurationType: 'connectionString',
connectionString: 'mongodb://localhost:27017',
})
.mockResolvedValueOnce({
configurationType: 'connectionString',
connectionString: 'mongodb://different-host:27017',
});
const client1 = await getMongoClient(mockContext, 1.1);
const client2 = await getMongoClient(mockContext, 1.1);
expect(MockMongoClient).toHaveBeenCalledTimes(2);
expect(MockMongoClient).toHaveBeenNthCalledWith(1, 'mongodb://localhost:27017', {
appName: 'devrel.integration.n8n_vector_integ',
driverInfo: {
name: 'n8n_vector',
version: '1.1',
},
});
expect(MockMongoClient).toHaveBeenNthCalledWith(2, 'mongodb://different-host:27017', {
appName: 'devrel.integration.n8n_vector_integ',
driverInfo: {
name: 'n8n_vector',
version: '1.1',
},
});
expect(mockClient1.connect).toHaveBeenCalledTimes(1);
expect(mockClient1.close).toHaveBeenCalledTimes(1);
expect(mockClient2.connect).toHaveBeenCalledTimes(1);
expect(mockClient2.close).not.toHaveBeenCalled();
expect(client1).toBe(mockClient1);
expect(client2).toBe(mockClient2);
expect(mockClient.connect).toHaveBeenCalledTimes(1);
expect(client).toBe(mockClient);
});
it('should create client with values configuration and port specified', async () => {
MockMongoClient.mockImplementation(() => mockClient1 as unknown as MongoClient);
MockMongoClient.mockImplementation(() => mockClient as unknown as MongoClient);
mockContext.getCredentials.mockResolvedValue({
configurationType: 'values',
host: 'localhost',
@@ -125,7 +95,7 @@ describe('VectorStoreMongoDBAtlas', () => {
database: 'testdb',
});
const client = await getMongoClient(mockContext, 1.1);
const client = await createMongoClient(mockContext, 1.1);
expect(MockMongoClient).toHaveBeenCalledTimes(1);
expect(MockMongoClient).toHaveBeenCalledWith('mongodb://testuser:testpass@localhost:27017', {
@@ -135,12 +105,12 @@ describe('VectorStoreMongoDBAtlas', () => {
version: '1.1',
},
});
expect(mockClient1.connect).toHaveBeenCalledTimes(1);
expect(client).toBe(mockClient1);
expect(mockClient.connect).toHaveBeenCalledTimes(1);
expect(client).toBe(mockClient);
});
it('should create client with values configuration without port (Atlas format)', async () => {
MockMongoClient.mockImplementation(() => mockClient1 as unknown as MongoClient);
MockMongoClient.mockImplementation(() => mockClient as unknown as MongoClient);
mockContext.getCredentials.mockResolvedValue({
configurationType: 'values',
host: 'cluster0.mongodb.net',
@@ -149,7 +119,7 @@ describe('VectorStoreMongoDBAtlas', () => {
database: 'atlasdb',
});
const client = await getMongoClient(mockContext, 1.1);
const client = await createMongoClient(mockContext, 1.1);
expect(MockMongoClient).toHaveBeenCalledTimes(1);
expect(MockMongoClient).toHaveBeenCalledWith(
@@ -162,92 +132,71 @@ describe('VectorStoreMongoDBAtlas', () => {
},
},
);
expect(mockClient1.connect).toHaveBeenCalledTimes(1);
expect(client).toBe(mockClient1);
expect(mockClient.connect).toHaveBeenCalledTimes(1);
expect(client).toBe(mockClient);
});
});
it('should reuse the same client when values configuration is unchanged', async () => {
MockMongoClient.mockImplementation(() => mockClient1 as unknown as MongoClient);
mockContext.getCredentials.mockResolvedValue({
configurationType: 'values',
host: 'localhost',
user: 'testuser',
password: 'testpass',
port: 27017,
database: 'testdb',
describe('.getCollections', () => {
const MockMongoClient = MongoClient as jest.MockedClass<typeof MongoClient>;
it('should create and close its own client', async () => {
const mockCollections = [{ name: 'Col1' }, { name: 'Col2' }];
const mockClient = {
connect: jest.fn().mockResolvedValue(undefined),
close: jest.fn().mockResolvedValue(undefined),
db: jest.fn().mockReturnValue({
listCollections: jest.fn().mockReturnValue({
toArray: jest.fn().mockResolvedValue(mockCollections),
}),
}),
};
MockMongoClient.mockImplementation(() => mockClient as unknown as MongoClient);
const context = mock<ILoadOptionsFunctions>({
getCredentials: jest.fn().mockResolvedValue({
configurationType: 'connectionString',
connectionString: 'mongodb://localhost:27017',
database: 'testdb',
}),
getNode: jest.fn().mockReturnValue({ typeVersion: 1.1 }),
});
const client1 = await getMongoClient(mockContext, 1.1);
const client2 = await getMongoClient(mockContext, 1.1);
const result = await getCollections.call(context);
expect(MockMongoClient).toHaveBeenCalledTimes(1);
expect(MockMongoClient).toHaveBeenCalledWith('mongodb://testuser:testpass@localhost:27017', {
appName: 'devrel.integration.n8n_vector_integ',
driverInfo: {
name: 'n8n_vector',
version: '1.1',
},
expect(result).toEqual({
results: [
{ name: 'Col1', value: 'Col1' },
{ name: 'Col2', value: 'Col2' },
],
});
expect(mockClient1.connect).toHaveBeenCalledTimes(1);
expect(mockClient1.close).not.toHaveBeenCalled();
expect(client1).toBe(mockClient1);
expect(client2).toBe(mockClient1);
expect(mockClient.connect).toHaveBeenCalledTimes(1);
expect(mockClient.close).toHaveBeenCalledTimes(1);
});
it('should create new client when values configuration changes', async () => {
MockMongoClient.mockImplementationOnce(
() => mockClient1 as unknown as MongoClient,
).mockImplementationOnce(() => mockClient2 as unknown as MongoClient);
mockContext.getCredentials
.mockResolvedValueOnce({
configurationType: 'values',
host: 'localhost',
user: 'testuser',
password: 'testpass',
port: 27017,
database: 'testdb',
})
.mockResolvedValueOnce({
configurationType: 'values',
host: 'different-host',
user: 'testuser',
password: 'testpass',
port: 27017,
database: 'testdb',
});
it('should close client even when an error occurs', async () => {
const mockClient = {
connect: jest.fn().mockResolvedValue(undefined),
close: jest.fn().mockResolvedValue(undefined),
db: jest.fn().mockReturnValue({
listCollections: jest.fn().mockReturnValue({
toArray: jest.fn().mockRejectedValue(new Error('db error')),
}),
}),
};
MockMongoClient.mockImplementation(() => mockClient as unknown as MongoClient);
const client1 = await getMongoClient(mockContext, 1.1);
const client2 = await getMongoClient(mockContext, 1.1);
const context = mock<ILoadOptionsFunctions>({
getCredentials: jest.fn().mockResolvedValue({
configurationType: 'connectionString',
connectionString: 'mongodb://localhost:27017',
database: 'testdb',
}),
getNode: jest.fn().mockReturnValue({ typeVersion: 1.1 }),
});
expect(MockMongoClient).toHaveBeenCalledTimes(2);
expect(MockMongoClient).toHaveBeenNthCalledWith(
1,
'mongodb://testuser:testpass@localhost:27017',
{
appName: 'devrel.integration.n8n_vector_integ',
driverInfo: {
name: 'n8n_vector',
version: '1.1',
},
},
);
expect(MockMongoClient).toHaveBeenNthCalledWith(
2,
'mongodb://testuser:testpass@different-host:27017',
{
appName: 'devrel.integration.n8n_vector_integ',
driverInfo: {
name: 'n8n_vector',
version: '1.1',
},
},
);
expect(mockClient1.connect).toHaveBeenCalledTimes(1);
expect(mockClient1.close).toHaveBeenCalledTimes(1);
expect(mockClient2.connect).toHaveBeenCalledTimes(1);
expect(mockClient2.close).not.toHaveBeenCalled();
expect(client1).toBe(mockClient1);
expect(client2).toBe(mockClient2);
await expect(getCollections.call(context)).rejects.toThrow('Error: db error');
expect(mockClient.close).toHaveBeenCalledTimes(1);
});
});
packages/@n8n/nodes-langchain/nodes/vector_store/VectorStoreMongoDBAtlas/VectorStoreMongoDBAtlas.node.ts
+30 -32
View File
@@ -149,50 +149,35 @@ const insertFields: INodeProperties[] = [
},
];
export const mongoConfig = {
client: null as MongoClient | null,
connectionString: '',
nodeVersion: 0,
};
/**
* Type used for cleaner, more intentional typing.
*/
type IFunctionsContext = IExecuteFunctions | ISupplyDataFunctions | ILoadOptionsFunctions;
/**
* Get the mongo client.
* Create a fresh MongoClient for the given context.
* Each call creates a new client to avoid connection pool sharing across executions.
* @param context - The context.
* @returns the MongoClient for the node.
* @param version - The node version.
* @returns A new, connected MongoClient.
*/
export async function getMongoClient(
export async function createMongoClient(
context: IExecuteFunctions | ISupplyDataFunctions | ILoadOptionsFunctions,
version: number,
) {
const credentials = await context.getCredentials(MONGODB_CREDENTIALS);
const node = context.getNode();
const { connectionString } = validateAndResolveMongoCredentials(node, credentials);
if (
!mongoConfig.client ||
mongoConfig.connectionString !== connectionString ||
mongoConfig.nodeVersion !== version
) {
if (mongoConfig.client) {
await mongoConfig.client.close();
}
mongoConfig.connectionString = connectionString;
mongoConfig.nodeVersion = version;
mongoConfig.client = new MongoClient(connectionString, {
appName: 'devrel.integration.n8n_vector_integ',
driverInfo: {
name: 'n8n_vector',
version: version.toString(),
},
});
await mongoConfig.client.connect();
}
return mongoConfig.client;
const client = new MongoClient(connectionString, {
appName: 'devrel.integration.n8n_vector_integ',
driverInfo: {
name: 'n8n_vector',
version: version.toString(),
},
});
await client.connect();
return client;
}
/**
@@ -211,8 +196,8 @@ export async function getDatabase(context: IFunctionsContext, client: MongoClien
* @returns The list of collections.
*/
export async function getCollections(this: ILoadOptionsFunctions) {
const client = await createMongoClient(this, this.getNode().typeVersion);
try {
const client = await getMongoClient(this, this.getNode().typeVersion);
const db = await getDatabase(this, client);
const collections = await db.listCollections().toArray();
const results = collections.map((collection) => ({
@@ -223,6 +208,8 @@ export async function getCollections(this: ILoadOptionsFunctions) {
return { results };
} catch (error) {
throw new NodeOperationError(this.getNode(), `Error: ${error.message}`);
} finally {
void client.close().catch(() => {});
}
}
@@ -276,16 +263,19 @@ export function getFilterValue<T>(
}
class ExtendedMongoDBAtlasVectorSearch extends MongoDBAtlasVectorSearch {
mongoClient: MongoClient;
preFilter: IDataObject;
postFilterPipeline?: IDataObject[];
constructor(
embeddings: EmbeddingsInterface,
options: MongoDBAtlasVectorSearchLibArgs,
mongoClient: MongoClient,
preFilter: IDataObject,
postFilterPipeline?: IDataObject[],
) {
super(embeddings, options);
this.mongoClient = mongoClient;
this.preFilter = preFilter;
this.postFilterPipeline = postFilterPipeline;
}
@@ -321,8 +311,8 @@ export class VectorStoreMongoDBAtlas extends createVectorStoreNode({
insertFields,
sharedFields,
async getVectorStoreClient(context, _filter, embeddings, itemIndex) {
const client = await createMongoClient(context, context.getNode().typeVersion);
try {
const client = await getMongoClient(context, context.getNode().typeVersion);
const db = await getDatabase(context, client);
const collectionName = getCollectionName(context, itemIndex);
const mongoVectorIndexName = getVectorIndexName(context, itemIndex);
@@ -357,10 +347,12 @@ export class VectorStoreMongoDBAtlas extends createVectorStoreNode({
textKey: metadataFieldName, // Field containing raw text
embeddingKey: embeddingFieldName, // Field containing embeddings
},
client,
preFilter ?? {},
postFilterPipeline,
);
} catch (error) {
void client.close().catch(() => {});
if (error instanceof NodeOperationError) {
throw error;
}
@@ -371,8 +363,8 @@ export class VectorStoreMongoDBAtlas extends createVectorStoreNode({
}
},
async populateVectorStore(context, embeddings, documents, itemIndex) {
const client = await createMongoClient(context, context.getNode().typeVersion);
try {
const client = await getMongoClient(context, context.getNode().typeVersion);
const db = await getDatabase(context, client);
const collectionName = getCollectionName(context, itemIndex);
const mongoVectorIndexName = getVectorIndexName(context, itemIndex);
@@ -396,6 +388,12 @@ export class VectorStoreMongoDBAtlas extends createVectorStoreNode({
itemIndex,
description: 'Please check your MongoDB Atlas connection details',
});
} finally {
void client.close().catch(() => {});
}
},
releaseVectorStoreClient(vectorStore) {
void vectorStore.mongoClient?.close().catch(() => {});
},
}) {}