Error `timeout of 20000ms exceeded` when you try to put files into maester

[if0s]

Bug Report

Description

When you try to put files larger than 1MB to storage using objectStorage.add method, there may be Error - timeout of 20000ms exceeded

Steps to Reproduce

To reproduce this, you can use following brunch of the example component and any public URL, for example https://psteam.pw/filebrowser/api/public/dl/RkT4tDlV/1.jpg

1. Create flow with any trigger
2. Use component above with provided URL (or any public URL with file larger than 1MB)
3. Save and run flow several times

Actual Result

Some of the executions will have Error timeout of 20000ms exceeded

Expected Result

No errors

Workaround(s)

none

[ArikLabs]

The problem looks as if at some point the stream stops writing to the mongodb, but at the same time the connection does not fall off. No runtime errors except StorageObjectTooLarge in redis, but it doesn't affect the write performance in mongodb.

Increasing the number of threads for event loops did not help (UV_THREADPOOL_SIZE=8/16/24).

The result is the same when using the master client and when directly through Axios.

Below is the code for local testing, both through the maeaster client and through axios

const ObjectStorage = require('@elastic.io/maester-client').ObjectStorage;
const axios = require('axios');
const jsonwebtoken = require("jsonwebtoken");
const fs = require("fs");

const token = jsonwebtoken.sign(
    {
        "tenantId": "56c207adb9121181e650c0ef",
        "contractId": "5b4f337bff4304610483ba67",
        "workspaceId": "59d341e9037f7200184a408b",
        "flowId": "*",
        "userId": "59d22e7eeb865b0018adc248"
    },
    "super-secret"
);

async function getStream(url) {
    try {
        url = 'https://psteam.pw/filebrowser/api/public/dl/RkT4tDlV/1.jpg';
        const response = await axios({
            method: 'GET',
            url,
            responseType: 'stream'
        });
        return response.data;
    } catch (error) {
        console.error(`Error fetching stream from URL: ${url}`);
        console.error(error.message);
        throw error;
    }
}

async function processAction() {
    try {
        const objectStorage = new ObjectStorage({
            jwtSecret: token,
            uri: 'http://localhost:3002',
            userAgent: 'example-component/3.0.0-dev.3 maester-client/5.0.2'
        });
        // console.log(objectStorage.client)
        const createdAttachmentId = await objectStorage.add(getStream, {
            headers: {},
            retryOptions: {
                requestTimeout: 30000,
                retriesCount: 2,
            },
        });

        console.log('Attachment created with ID:', createdAttachmentId);
    } catch (error) {
        console.error('Error processing action:');
        console.error(error.message);
    }
}
async function delay(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
}

for (let i = 0; i < 50; i++) {
    // await delay(1000);
    processAction();
}

const axios = require('axios');
const jsonwebtoken = require("jsonwebtoken")
const fs = require("fs");

async function getStream(url) {
    const getStream = async () => {
        const fileStream = await axios({
            method: 'GET',
            url,
            responseType: 'stream'
        });

        return fileStream;
    };

    const stream = await getStream();
    return stream.data;
}

async function uploadStream(stream, endpointUrl, headers) {
    const api = axios.create();
    const axiosReqConfig = {
        method: 'post',
        url: endpointUrl,
        headers: {
            ...headers,
        }};
    try {
       const response = await api.request(
            {
                ...axiosReqConfig,
                data: stream,
                timeout: 30000
            }
       )
        console.log(response.data)
    } catch (error) {
        if (error.response) {
            console.error('Error:', error.response.status, error.response.data);
        } else {
            console.error('Error:', error.message);
        }
    }
}

async function main() {
    const fileUrl = 'https://psteam.pw/filebrowser/api/public/dl/RkT4tDlV/1.jpg';
    const endpointUrl = 'http://localhost:3002/objects';


    const token = jsonwebtoken.sign(
        {
            "tenantId": "56c207adb9121181e650c0ef",
            "contractId": "5b62c91bfd98ea00112d5668",
            "workspaceId": "59d34d9af1ffe00019df5f3f",
            "flowId": "*",
            "userId": "59d22e7eeb865b0018adc248"
        },
        "super-secret"
    );


    const headers = {
        'X-Query-key1': 'value1',
        'X-Meta-key2': 'value2',
        'Authorization': `Bearer ${token}`
    };

    const stream = await getStream(fileUrl);

    console.log('Got file data stream, start uploading to internal storage')

    await uploadStream(stream, endpointUrl, headers);
}

async function delay(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
}

async function start() {
  for (let i = 0; i < 1; i++){
    // await delay(20);
    main();
  }
}

start();