From 87ca22ce9c970af99771a2257c3b234c861e9ba7 Mon Sep 17 00:00:00 2001 From: Eric Peterson Date: Fri, 7 Apr 2023 14:59:03 +0200 Subject: [PATCH 1/5] Close transaction on upstream error Signed-off-by: Eric Peterson --- .changeset/search-destroyer-of-worlds.md | 7 ++++ .../src/engines/ElasticSearchSearchEngine.ts | 3 ++ .../PgSearchEngineIndexer.test.ts | 34 +++++++++++++++++++ .../PgSearchEngine/PgSearchEngineIndexer.ts | 28 +++++++++++++++ 4 files changed, 72 insertions(+) create mode 100644 .changeset/search-destroyer-of-worlds.md diff --git a/.changeset/search-destroyer-of-worlds.md b/.changeset/search-destroyer-of-worlds.md new file mode 100644 index 0000000000..e5d29d2ad1 --- /dev/null +++ b/.changeset/search-destroyer-of-worlds.md @@ -0,0 +1,7 @@ +--- +'@backstage/plugin-search-backend-module-pg': patch +--- + +Fixed a bug that could cause orphaned PG connections to accumulate (eventually +exhausting available connections) when errors were encountered earlier in the +search indexing process. diff --git a/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngine.ts b/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngine.ts index 25df3101de..3294ca9d33 100644 --- a/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngine.ts +++ b/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngine.ts @@ -285,6 +285,9 @@ export class ElasticSearchSearchEngine implements SearchEngine { }); // Attempt cleanup upon failure. + // todo(@backstage/discoverability-maintainers): Consider introducing a more + // formal mechanism for handling such errors in BatchSearchEngineIndexer and + // replacing this handler with it. See: #17291 indexer.on('error', async e => { indexerLogger.error(`Failed to index documents for type ${type}`, e); let cleanupError: Error | undefined; diff --git a/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts b/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts index f1cb3923e1..d6369d4ac3 100644 --- a/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts +++ b/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts @@ -15,6 +15,7 @@ */ import { TestPipeline } from '@backstage/plugin-search-backend-node'; import { range } from 'lodash'; +import { Transform } from 'stream'; import { PgSearchEngineIndexer } from './PgSearchEngineIndexer'; import { DatabaseStore } from '../database'; @@ -156,4 +157,37 @@ describe('PgSearchEngineIndexer', () => { expect(result.error).toBe(expectedError); expect(tx.rollback).toHaveBeenCalledWith(expectedError); }); + + it('should rollback transaction on upstream error', async () => { + // Given a decorator that results in an error + let counter = 0; + const expectedError = new Error('Upstream error'); + const errorDecorator = new Transform({ objectMode: true }); + errorDecorator._transform = (chunk, _enc, cb) => { + counter++; + if (counter > 1) { + cb(expectedError); + } else { + cb(undefined, chunk); + } + }; + + // When the decorator is run in a pipeline with the PG indexer + const result = await TestPipeline.fromIndexer(indexer) + .withDecorator(errorDecorator) + .withDocuments([ + { title: 'a', text: 'a', location: '/a' }, + { title: 'b', text: 'b', location: '/b' }, + ]) + .execute(); + + // And we allow async teardown logic to complete + await new Promise(resolve => setImmediate(resolve)); + + // Then the transaction should have been closed with the expected error. + expect(database.getTransaction).toHaveBeenCalledTimes(1); + expect(database.completeInsert).not.toHaveBeenCalled(); + expect(result.error).toBe(expectedError); + expect(tx.rollback).toHaveBeenCalledWith(expectedError); + }); }); diff --git a/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.ts b/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.ts index ba625bc116..d1ff771d40 100644 --- a/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.ts +++ b/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.ts @@ -92,4 +92,32 @@ export class PgSearchEngineIndexer extends BatchSearchEngineIndexer { throw e; } } + + /** + * Custom handler covering the case where an error occurred somewhere else in + * the indexing pipeline (e.g. a collator or decorator). In such cases, the + * finalize method is not called, which leaves a dangling transaction and + * therefore an open connection to PG. This handler ensures we close the + * transaction and associated connection. + * + * todo(@backstage/discoverability-maintainers): Consider introducing a more + * formal mechanism for handling such errors in BatchSearchEngineIndexer and + * replacing this method with it. See: #17291 + * + * @internal + */ + async _destroy(error: Error | null, done: (error?: Error | null) => void) { + // Ignore situations where there was no error. + if (!error) { + done(); + return; + } + + try { + this.tx!.rollback(error); + } catch { + // Unlikely! It was likely rolled back earlier. + } + done(); + } } From 8bcbe272151bb0e9d39709029d8ee6405f27383c Mon Sep 17 00:00:00 2001 From: Eric Peterson Date: Fri, 7 Apr 2023 15:52:37 +0200 Subject: [PATCH 2/5] Fix tests on node v16 Signed-off-by: Eric Peterson --- .../src/PgSearchEngine/PgSearchEngineIndexer.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.ts b/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.ts index d1ff771d40..a0ad3ee7d9 100644 --- a/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.ts +++ b/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.ts @@ -118,6 +118,6 @@ export class PgSearchEngineIndexer extends BatchSearchEngineIndexer { } catch { // Unlikely! It was likely rolled back earlier. } - done(); + done(error); } } From acb9ae3672f1d918c54c2660d156a1e6888cc37e Mon Sep 17 00:00:00 2001 From: Eric Peterson Date: Mon, 10 Apr 2023 11:11:20 +0200 Subject: [PATCH 3/5] Assert that rollback is not attempted when there are no errors. Signed-off-by: Eric Peterson --- .../src/PgSearchEngine/PgSearchEngineIndexer.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts b/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts index d6369d4ac3..a8e010dea5 100644 --- a/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts +++ b/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts @@ -65,6 +65,7 @@ describe('PgSearchEngineIndexer', () => { ); expect(database.completeInsert).toHaveBeenCalledWith(tx, 'my-type'); expect(tx.commit).toHaveBeenCalled(); + expect(tx.rollback).not.toHaveBeenCalled(); }); it('should batch insert documents', async () => { From b4bbf9e18e8a1faff48c81c349e633d531306f94 Mon Sep 17 00:00:00 2001 From: Eric Peterson Date: Mon, 10 Apr 2023 11:12:57 +0200 Subject: [PATCH 4/5] Check that transaction is not completed before attempting rollback Signed-off-by: Eric Peterson --- .../src/PgSearchEngine/PgSearchEngineIndexer.test.ts | 2 ++ .../src/PgSearchEngine/PgSearchEngineIndexer.ts | 7 +++---- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts b/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts index a8e010dea5..d614185b7c 100644 --- a/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts +++ b/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts @@ -23,6 +23,7 @@ describe('PgSearchEngineIndexer', () => { const tx = { rollback: jest.fn(), commit: jest.fn(), + isCompleted: jest.fn(), } as any; let database: jest.Mocked; let indexer: PgSearchEngineIndexer; @@ -37,6 +38,7 @@ describe('PgSearchEngineIndexer', () => { completeInsert: jest.fn(), prepareInsert: jest.fn(), }; + tx.isCompleted.mockReturnValue(false); indexer = new PgSearchEngineIndexer({ batchSize: 100, type: 'my-type', diff --git a/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.ts b/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.ts index a0ad3ee7d9..47ea812a92 100644 --- a/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.ts +++ b/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.ts @@ -113,11 +113,10 @@ export class PgSearchEngineIndexer extends BatchSearchEngineIndexer { return; } - try { - this.tx!.rollback(error); - } catch { - // Unlikely! It was likely rolled back earlier. + if (!this.tx!.isCompleted()) { + await this.tx!.rollback(error); } + done(error); } } From 719602451e5f57a8959297fd0439d8af882d97c6 Mon Sep 17 00:00:00 2001 From: Eric Peterson Date: Mon, 10 Apr 2023 11:13:36 +0200 Subject: [PATCH 5/5] Test that rollback is only attempted once in all cases. Signed-off-by: Eric Peterson --- .../src/PgSearchEngine/PgSearchEngineIndexer.test.ts | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts b/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts index d614185b7c..37af4f00f0 100644 --- a/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts +++ b/plugins/search-backend-module-pg/src/PgSearchEngine/PgSearchEngineIndexer.test.ts @@ -91,7 +91,7 @@ describe('PgSearchEngineIndexer', () => { expect(database.getTransaction).toHaveBeenCalledTimes(1); expect(database.insertDocuments).not.toHaveBeenCalled(); expect(database.completeInsert).not.toHaveBeenCalled(); - expect(tx.rollback).toHaveBeenCalled(); + expect(tx.rollback).toHaveBeenCalledTimes(1); }); it('should close out stream and bubble up error on prepare', async () => { @@ -104,6 +104,7 @@ describe('PgSearchEngineIndexer', () => { }, ]; + tx.isCompleted.mockReturnValue(true); database.prepareInsert.mockRejectedValueOnce(expectedError); const result = await TestPipeline.fromIndexer(indexer) .withDocuments(documents) @@ -113,6 +114,7 @@ describe('PgSearchEngineIndexer', () => { expect(database.insertDocuments).not.toHaveBeenCalled(); expect(database.completeInsert).not.toHaveBeenCalled(); expect(result.error).toBe(expectedError); + expect(tx.rollback).toHaveBeenCalledTimes(1); expect(tx.rollback).toHaveBeenCalledWith(expectedError); }); @@ -126,6 +128,7 @@ describe('PgSearchEngineIndexer', () => { }, ]; + tx.isCompleted.mockReturnValue(true); database.insertDocuments.mockRejectedValueOnce(expectedError); const result = await TestPipeline.fromIndexer(indexer) .withDocuments(documents) @@ -135,6 +138,7 @@ describe('PgSearchEngineIndexer', () => { expect(database.prepareInsert).toHaveBeenCalledTimes(1); expect(database.completeInsert).not.toHaveBeenCalled(); expect(result.error).toBe(expectedError); + expect(tx.rollback).toHaveBeenCalledTimes(1); expect(tx.rollback).toHaveBeenCalledWith(expectedError); }); @@ -148,6 +152,7 @@ describe('PgSearchEngineIndexer', () => { }, ]; + tx.isCompleted.mockReturnValue(true); database.completeInsert.mockRejectedValueOnce(expectedError); const result = await TestPipeline.fromIndexer(indexer) .withDocuments(documents) @@ -158,6 +163,7 @@ describe('PgSearchEngineIndexer', () => { expect(database.insertDocuments).toHaveBeenCalledTimes(1); expect(database.completeInsert).toHaveBeenCalledTimes(1); expect(result.error).toBe(expectedError); + expect(tx.rollback).toHaveBeenCalledTimes(1); expect(tx.rollback).toHaveBeenCalledWith(expectedError); }); @@ -191,6 +197,7 @@ describe('PgSearchEngineIndexer', () => { expect(database.getTransaction).toHaveBeenCalledTimes(1); expect(database.completeInsert).not.toHaveBeenCalled(); expect(result.error).toBe(expectedError); + expect(tx.rollback).toHaveBeenCalledTimes(1); expect(tx.rollback).toHaveBeenCalledWith(expectedError); }); });