From 33391e6e151b2d50c75d0c056992c43bd73144a4 Mon Sep 17 00:00:00 2001 From: Albert Alises Date: Wed, 11 Mar 2026 16:35:12 +0100 Subject: [PATCH] fix(MongoDBVectorStore Node): Replace MongoDB Atlas Vector Store singleton with per-execution client (#26906) --- .../VectorStoreMongoDBAtlas.node.test.ts | 237 +++++++----------- .../VectorStoreMongoDBAtlas.node.ts | 62 +++-- 2 files changed, 123 insertions(+), 176 deletions(-) diff --git a/packages/@n8n/nodes-langchain/nodes/vector_store/VectorStoreMongoDBAtlas/VectorStoreMongoDBAtlas.node.test.ts b/packages/@n8n/nodes-langchain/nodes/vector_store/VectorStoreMongoDBAtlas/VectorStoreMongoDBAtlas.node.test.ts index ab30f1c2bbb..593908d6992 100644 --- a/packages/@n8n/nodes-langchain/nodes/vector_store/VectorStoreMongoDBAtlas/VectorStoreMongoDBAtlas.node.test.ts +++ b/packages/@n8n/nodes-langchain/nodes/vector_store/VectorStoreMongoDBAtlas/VectorStoreMongoDBAtlas.node.test.ts @@ -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({ 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; - 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; + + 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({ + 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({ + 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); }); }); diff --git a/packages/@n8n/nodes-langchain/nodes/vector_store/VectorStoreMongoDBAtlas/VectorStoreMongoDBAtlas.node.ts b/packages/@n8n/nodes-langchain/nodes/vector_store/VectorStoreMongoDBAtlas/VectorStoreMongoDBAtlas.node.ts index a322fded9a4..7c9a0134dea 100644 --- a/packages/@n8n/nodes-langchain/nodes/vector_store/VectorStoreMongoDBAtlas/VectorStoreMongoDBAtlas.node.ts +++ b/packages/@n8n/nodes-langchain/nodes/vector_store/VectorStoreMongoDBAtlas/VectorStoreMongoDBAtlas.node.ts @@ -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( } 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(() => {}); + }, }) {}