nodejs – Porque a veces no se obtienen datos del fichero csv


Tengo un problema, y es que necesito obtener un fichero csv a traves de FTP. Conecto, descargo el fichero, lo creo escribiendo en el con fs.createWriteStream. Una vez finalizado este proceso, procedo a leerlo con fs.createReadStream y lo “parseo” con la dependencia csv-parser, de tal forma que pueda obtener esos datos en un array de objetos (cada una de esas lineas). El caso es que a veces obtengo los datos, y otras veces no. Creo que es porque no estoy manejando los eventos asincrónicos como es debido, pero no conigo enteneder donde estoy fallando. Sería de gran ayuda si alguien puediera aclararmelo. Por aquí adjunto el código:

async function getStockAndPrice(req, res, next) {
    try {
        let options = {
            host: 'xxxxxx',
            user: 'xxxxx',
            password: 'xxxxx',
            port: 21,
            secure: true,
            secureOptions: { rejectUnauthorized: false }
        }
        let filesFromFTP = await getFileFromFTP(options, 'stock.csv', 'dataFromFTP.csv');
        let productFilePath = filesFromFTP(0).path;
        let stream = fs.createReadStream(productFilePath);
        let productsData = await getProductsFromFile(stream);
        return res.send({ message: "Artículos obtenidos con éxito.", result: productsData });
    } catch (e) {
        next(e);
    }
}



async function getProductsFromFile(stream) {
    return new Promise(resolve => {
        let allProducts = ();
        stream.pipe(csv({ headers: ('ean', 'stock', 'price'), separator: ";" }))
            .on('data', (row) => {
                console.log(row);
                allProducts.push(row);
            })
            .on('end', () => {
                console.log(allProducts);
                resolve(allProducts);
            });
    });
}


async function getFileFromFTP(options, fileName, newFileName) {
    try {
        let c = new ftp();
        c.connect(options);
        c.listAsync = bluebird.promisify(c.list);
        c.getAsync = bluebird.promisify(c.get);
        let list = await c.listAsync().filter(f => f.name === fileName);
        if (!list.length) {
            throw new Error("No se ha encontrado el fichero.");
        }
        let downloadedFiles = await bluebird.mapSeries(list, async le => {
            return await c.getAsync(le.name);
        });
        c.end();
        let writtenFiles = downloadedFiles.map(df => {
            return df.pipe(fs.createWriteStream(newFileName, { encoding: 'utf8' }));
        });
        return writtenFiles;
    } catch (e) {
        throw new Error(e.message);
    }
}

Creo que el fallo principal está en los streams de lectura y escritura, que no llegan a terminar a tiempo, y a veces el fichero que se lee no tiene los datos. Alguna idea? No se si debería documnetar algo más a mayores, pero estas son las funciones con las que realizo el procedimiento. Quiza no lo tengo bien entendido. Gracias de antemano.