Question memory usage cluster mode , sandboxed processes

[smartdev10]

Thank you for this great library

Can you please advise me if this is good implementation

I'm trying to processes a list of emails of 200k each on it's own process , i am using puppeteer , could be multiple Queues
I create Queues on the fly using socket.io that creates a worker_thread that create the Queue this is my code

The processor open a headless browser and preforms some operations and closes and i do this concurrently

The Question is how should i add jobs to the Queue , now i add them all at once is this good?

How should i approach this ?

I noticed memory usage keeps increasing over time i don't know why

this is my code

thank you

import { workerData , parentPort } from 'worker_threads'
import Queue, { Job } from 'bull';
import Emmitter from 'socket.io-emitter'
import os from "os"

const numCpus = os.cpus().length
const io = Emmitter({ host:process.env.REDIS_HOST , port:6379 })

const { actions , emails , compain , port } = workerData
    
const queue = new Queue(String(compain),{
    redis:{
        host:process.env.REDIS_HOST,
        port:6379
    },
   settings:{
     maxStalledCount:0,
     stalledInterval:0
   }
})

parentPort.on('message',(value)=>{
    if(value === 'stop'){
      queue.pause()
      .then(() => queue.empty())
      .then(() => queue.clean(0))
      .then(() => queue.close())
      .then(() => {
        parentPort.postMessage({ message:"stopped" , data: {compain} })
        // io.emit(`stop_done_${compain}`)
      }).catch(()=>{
        parentPort.postMessage({ message:"stopped" , data: {compain} })
        // io.emit(`stopped_ready_${compain}`)
      })
    }
    if(value === 'pause'){
      console.log('pause')
        queue.pause(true,true).then(() => {
          parentPort.postMessage({ message:"paused" , data: {compain} })
          io.emit(`pause_done_${compain}`)
        })
    }
    if(value === 'resume'){
      queue.resume(true).then(() => {
         parentPort.postMessage({ message:"resumed" , data: {compain} })
         io.emit(`resume_done_${compain}`)
      })
    }
})
 
queue.empty()
.then(() => queue.clean(0))
.then(() => {
    let jobs = emails.map((seed:any)=> {
        let proxy = 'null'
        if(seed.proxy !== 'null'){
            proxy = seed.proxy
        }
        return { 
            data : { email:seed.email , password:seed.password , proxy , actions , seedId:seed.id , compain },
            opts:  { attempts:0 , removeOnComplete :true , removeOnFail:true }
        }
    })
   return queue.addBulk(jobs)
}).then(()=> queue.process("*" , numCpus , __dirname + '/worker.processor.js'))
    

queue.on("failed" , (job,err) => {
    console.log(err)
    parentPort.postMessage({ message:"failed" , data:{...job.data} })
    io.emit(`failed_${job.data.seedId}`)
    job.discard().then(() => { 
      console.log(`Job ${job.id} of queue ${job.id} failed with error ${err.message}`)
    })
})

queue.on("resumed" , (job:Job) => {
  console.log(job);
})


queue.on("progress" , (job : Job , progress) => {
  console.log(`Job ${job.id} has a feedback of ${progress}`);
  parentPort.postMessage({ message:"screenshot" , seedId:job.data.seedId , path:progress.path })
})


queue.on("drained" , () => {
  queue.pause(true,false)
  .then(() => queue.resume())
  .then(()=> parentPort.postMessage({ message:"drained" , data: {compain} }))
})


queue.on("completed" , (job , result) => {
  parentPort.postMessage({ message:"completed" , data:{...job.data} })
  io.emit(`done_${job.data.seedId}`)
  job.remove()
})

queue.on("active" , (job) => {
  parentPort.postMessage({ message:"active" , data:{...job.data} })
  io.emit(`active_${job.data.seedId}`)
 })

queue.on("stalled" , (job) => {
  job.remove().then(()=>{
    parentPort.postMessage({ message:"creashed" , data:{...job.data} })
    console.log(`Job ${job.id} creashed`);
  })
})