[7.17] Handle content stream errors in report pre-deletion (#173792) (#174141)

# Backport This will backport the following commits from `main` to `7.17`: - [Handle content stream errors in report pre-deletion (#173792)](https://github.com/elastic/kibana/pull/173792)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sqren/backport)
2025-04-24 17:59:23 -04:00 · 2024-01-04 11:38:18 -07:00 · 2024-01-04 11:38:18 -07:00 · eb687c43d6
commit eb687c43d6
parent 07dc7a1a9f
3 changed files with 66 additions and 8 deletions
--- a/x-pack/plugins/reporting/server/routes/lib/job_response_handler.test.ts
+++ b/x-pack/plugins/reporting/server/routes/lib/job_response_handler.test.ts
@ -99,7 +99,7 @@ describe('deleteJobResponseHandler', () => {
      ReturnType<typeof jobsQuery.get>
    >);
    jobsQuery.delete.mockRejectedValueOnce(
-      Object.assign(new Error('Some error.'), { statusCode: 123 })
+      Object.assign(new Error('Some error.'), { statusCode: 500 })
    );
    await deleteJobResponseHandler(
      core,
@ -110,8 +110,7 @@ describe('deleteJobResponseHandler', () => {
    );

    expect(response.customError).toHaveBeenCalledWith({
-      statusCode: 123,
-      body: 'Some error.',
+      statusCode: 500,
    });
  });
 });
--- a/x-pack/plugins/reporting/server/routes/lib/job_response_handler.ts
+++ b/x-pack/plugins/reporting/server/routes/lib/job_response_handler.ts
@ -5,7 +5,6 @@
 * 2.0.
 */

-import { promisify } from 'util';
 import { kibanaResponseFactory } from 'kibana/server';
 import { ReportingCore } from '../../';
 import { ALLOWED_JOB_CONTENT_TYPES } from '../../../common/constants';
@ -90,18 +89,41 @@ export async function deleteJobResponseHandler(

  const docIndex = doc.index;
  const stream = await getContentStream(reporting, { id: docId, index: docIndex });
+  const reportingSetup = reporting.getPluginSetupDeps();
+  const logger = reportingSetup.logger.clone(['delete-report']);
+
+  // An "error" event is emitted if an error is
+  // passed to the `stream.end` callback from
+  // the _final method of the ContentStream.
+  // This event must be handled.
+  stream.on('error', (err) => {
+    logger.error(err);
+  });

  try {
-    /** @note Overwriting existing content with an empty buffer to remove all the chunks. */
-    await promisify(stream.end.bind(stream, '', 'utf8'))();
+    // Overwriting existing content with an
+    // empty buffer to remove all the chunks.
+    await new Promise<void>((resolve, reject) => {
+      stream.end('', 'utf8', (error?: Error) => {
+        if (error) {
+          // handle error that could be thrown
+          // from the _write method of the ContentStream
+          reject(error);
+        } else {
+          resolve();
+        }
+      });
+    });
+
    await jobsQuery.delete(docIndex, docId);
+
    return res.ok({
      body: { deleted: true },
    });
  } catch (error) {
+    logger.error(error);
    return res.customError({
-      statusCode: error.statusCode,
-      body: error.message,
+      statusCode: 500,
    });
  }
 }
--- a/x-pack/plugins/reporting/server/routes/management/jobs.test.ts
+++ b/x-pack/plugins/reporting/server/routes/management/jobs.test.ts
@ -363,4 +363,41 @@ describe('GET /api/reporting/jobs/download', () => {
        );
    });
  });
+
+  describe('delete report', () => {
+    const getCompleteHits = ({
+      jobType = 'unencodedJobType',
+      outputContentType = 'text/plain',
+      title = '',
+    } = {}) => {
+      return getHits({
+        jobtype: jobType,
+        status: 'completed',
+        output: { content_type: outputContentType },
+        payload: { title },
+      });
+    };
+
+    it('handles content stream errors', async () => {
+      stream = new Readable({
+        read() {
+          this.push('test');
+          this.push(null);
+        },
+      }) as typeof stream;
+      stream.end = jest.fn().mockImplementation((_name, _encoding, callback) => {
+        callback(new Error('An error occurred in ending the content stream'));
+      });
+
+      (getContentStream as jest.MockedFunction<typeof getContentStream>).mockResolvedValue(stream);
+      mockEsClient.search.mockResolvedValueOnce({ body: getCompleteHits() } as any);
+      registerJobInfoRoutes(core);
+
+      await server.start();
+      await supertest(httpSetup.server.listener)
+        .delete('/api/reporting/jobs/delete/denk')
+        .expect(500)
+        .expect('Content-Type', 'application/json; charset=utf-8');
+    });
+  });
 });