Merge pull request #2153 from Inist-CNRS/2136-erreur-import-gros-fich…

…iers catch the right error and avoid cancelling and deleting the entire da…
Inist-CNRS · Sep 18, 2024 · ecc7327 · ecc7327
2 parents c4f56c2 + e3a7686
commit ecc7327
Show file tree

Hide file tree

Showing 8 changed files with 96 additions and 123 deletions.
diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -75,9 +75,9 @@
         "@emotion/react": "11.7.1",
         "@emotion/styled": "11.6.0",
         "@ezs/analytics": "2.3.2",
-        "@ezs/basics": "2.7.1",
+        "@ezs/basics": "2.7.2",
         "@ezs/conditor": "2.12.2",
-        "@ezs/core": "3.10.3",
+        "@ezs/core": "3.10.4",
         "@ezs/istex": "1.5.9",
         "@ezs/lodex": "file:./packages/ezsLodex",
         "@ezs/sparql": "1.2.3",
@@ -144,7 +144,7 @@
         "mongodb": "4.17.1",
         "mongodb-restore": "1.6.2",
         "mui-file-dropzone": "4.0.2",
-        "multistream": "^2.1.1",
+        "multistream": "4.1.0",
         "node-polyglot": "^2.3.1",
         "pdfkit": "0.14.0",
         "pm2": "5.3.1",

diff --git a/src/api/controller/api/upload.js b/src/api/controller/api/upload.js
@@ -3,14 +3,14 @@ import route from 'koa-route';
 import asyncBusboy from '@recuperateur/async-busboy';
 import config from 'config';
 import koaBodyParser from 'koa-bodyparser';
-import fs from 'fs';
 
 import progress from '../../services/progress';
 import { PENDING, UPLOADING_DATASET } from '../../../common/progressStatus';
 import {
     saveStreamInFile,
     checkFileExists,
     getUploadedFileSize,
+    unlinkFile,
 } from '../../services/fsHelpers';
 
 import { v1 as uuid } from 'uuid';
@@ -19,13 +19,8 @@ import { IMPORT } from '../../workers/import';
 
 export const requestToStream = (asyncBusboyImpl) => async (req) => {
     const { files, fields } = await asyncBusboyImpl(req);
-
-    files[0].once('close', () => {
-        try {
-            fs.unlinkSync(files[0].path);
-        } catch (error) {
-            console.warn(error);
-        }
+    files[0].once('close', async () => {
+        await unlinkFile(files[0].path);
     });
     return { stream: files[0], fields };
 };

diff --git a/src/api/services/fsHelpers.js b/src/api/services/fsHelpers.js
@@ -1,7 +1,7 @@
 import fs from 'fs';
 import range from 'lodash/range';
 import rangeRight from 'lodash/rangeRight';
-import multiStream from 'multistream';
+import MultiStream from 'multistream';
 
 import composeAsync from '../../common/lib/composeAsync';
 import safePipe from './safePipe';
@@ -29,15 +29,15 @@ export const createWriteStream = (chunkname) => fs.createWriteStream(chunkname);
 export const createReadStream = (chunkname) => fs.createReadStream(chunkname);
 
 export const mergeChunksFactory =
-    (createReadStreamImpl, multiStreamImpl) => (filename, nbChunks) => {
+    (createReadStreamImpl) => (filename, nbChunks) => {
         const sourceStreams = range(1, nbChunks + 1)
             .map((nb) => `${filename}.${nb}`)
             .map((chunkname) => createReadStreamImpl(chunkname));
 
-        return multiStreamImpl(sourceStreams);
+        return new MultiStream(sourceStreams);
     };
 
-export const mergeChunks = mergeChunksFactory(createReadStream, multiStream);
+export const mergeChunks = mergeChunksFactory(createReadStream);
 
 export const getFileStats = (filename) =>
     new Promise((resolve, reject) => {

diff --git a/src/api/services/fsHelpers.spec.js b/src/api/services/fsHelpers.spec.js
@@ -99,29 +99,4 @@ describe('fsHelpers', () => {
             });
         });
     });
-
-    describe('mergeChunksFactory', () => {
-        const createReadStreamImpl = jest.fn((v) => `read stream for ${v}`);
-        const multiStreamImpl = jest.fn(() => 'merged stream');
-        beforeAll(async () => {
-            await mergeChunksFactory(createReadStreamImpl, multiStreamImpl)(
-                'filename',
-                3,
-            );
-        });
-
-        it('should have called createReadStreamImpl with each generated chunkname', () => {
-            expect(createReadStreamImpl).toHaveBeenCalledWith('filename.1');
-            expect(createReadStreamImpl).toHaveBeenCalledWith('filename.2');
-            expect(createReadStreamImpl).toHaveBeenCalledWith('filename.3');
-        });
-
-        it('should have called multiStreamImpl with created array of readStream and write stream', () => {
-            expect(multiStreamImpl).toHaveBeenCalledWith([
-                'read stream for filename.1',
-                'read stream for filename.2',
-                'read stream for filename.3',
-            ]);
-        });
-    });
 });
diff --git a/src/api/services/import.js b/src/api/services/import.js
@@ -14,24 +14,25 @@ import localConfig from '../../../config.json';
 
 ezs.use(ezsBasics);
 
-export const getLoader = (loaderName, loaderEnvironment, loaderHeader) => (stream) => {
-    const env2query = new URLSearchParams(loaderEnvironment);
-    return stream
-        .pipe(
-            ezs('URLConnect', {
-                url: `${
-                    process.env.WORKERS_URL || 'http://localhost:31976'
-                }/loaders/${loaderName}?${env2query}`,
-                streaming: true,
-                timeout: Number(localConfig.timeout) || 120000,
-                json: false,
-                encoder: 'transit',
-                header: loaderHeader,
-            }),
-        )
-        .pipe(ezs('unpack'));
-};
-export const getCustomLoader = async (script, loaderEnvironment) => {
+export const getLoader =
+    (loaderName, loaderEnvironment, loaderHeader) => (stream) => {
+        const env2query = new URLSearchParams(loaderEnvironment);
+        return stream
+            .pipe(
+                ezs('URLConnect', {
+                    url: `${
+                        process.env.WORKERS_URL || 'http://localhost:31976'
+                    }/loaders/${loaderName}?${env2query}`,
+                    streaming: true,
+                    timeout: Number(localConfig.timeout) || 120000,
+                    json: false,
+                    encoder: 'transit',
+                    header: loaderHeader,
+                }),
+            )
+            .pipe(ezs('unpack'));
+    };
+export const getCustomLoader = (script, loaderEnvironment) => {
     return (stream) =>
         stream.pipe(ezs('delegate', { script }, loaderEnvironment));
 };
@@ -86,37 +87,38 @@ export const startImport = async (ctx) => {
         if (customLoader) {
             loaderEnvironment.parser =
                 loaderEnvironment.parser.concat('/custom');
-            parseStream = await ctx.getCustomLoader(
-                customLoader,
+            parseStream = ctx.getCustomLoader(customLoader, loaderEnvironment);
+        } else {
+            parseStream = ctx.getLoader(
+                parser,
                 loaderEnvironment,
+                `'X-Request-ID:${fusible}`,
             );
-        } else {
-            parseStream = ctx.getLoader(parser, loaderEnvironment, `'X-Request-ID:${fusible}`);
         }
         let stream;
         if (url) {
             stream = await ctx.getStreamFromUrl(url);
         }
         if (filename && totalChunks) {
-            try {
-                stream = ctx.mergeChunks(filename, totalChunks);
-            } catch (error) {
-                throw new Error(`Error while merging chunks: ${error}`);
-            }
+            stream = ctx.mergeChunks(filename, totalChunks);
         }
         if (text) {
             loaderEnvironment.source = 'text input';
             stream = ctx.getStreamFromText(text);
         }
         const inputStream = stream.pipe(ezs(breaker, { fusible }));
-        const parsedStream = await parseStream(inputStream);
+        const parsedStream = parseStream(inputStream);
         const outputStream = parsedStream.pipe(ezs(breaker, { fusible }));
-        await ctx.saveParsedStream(ctx, outputStream);
+
+        const insertedTotal = await ctx.saveParsedStream(ctx, outputStream);
         progress.start(ctx.tenant, {
             status: INDEXATION,
             type: 'import',
         });
         await ctx.dataset.indexColumns();
+    } catch (error) {
+        console.error('Error during import', error);
+        throw new Error(`Error during import: ${error}`);
     } finally {
         await disableFusible(fusible);
         progress.finish(ctx.tenant);

diff --git a/src/api/services/import.spec.js b/src/api/services/import.spec.js
@@ -107,7 +107,7 @@ describe.skip('import', () => {
                 },
                 update: jest.fn(),
             },
-            getCustomLoader: jest.fn().mockImplementation(() => loader),
+            getCustomLoader: () => loader,
             mergeChunks: jest.fn().mockImplementation(() => 'stream'),
             clearChunks: jest.fn(),
             saveParsedStream: jest.fn(),