Having a weird issue with hworker and a mailing list

[naglalakk]

Hi. This is probably not a bug or anything but I'm just wondering what I'm doing wrong in this case.
I have an email list with over 3000 people. At first I was triggering a single job that would take all the emails in a single mailing list and send out a single email to all of them. Because this job was fairly big (gathering all emails in a single email list with about 3000-5000 emails in the list) we ran into problems with the timeout which I didn't notice at first. The default timeout is 120 seconds. I bumped this up to half an hour so hworker would definitely not think that my job had timed out and start sending emails again. This did not work however and after half an hour the worker started sending emails again. Maybe my calculations were wrong and the task of sending all these emails was longer then 30 minutes.

But then I decided to take different approach. A single job is now finding all the emails in the database and creating a new job for each. So the worker gets a job for each email being sent to, sends the email and then moves on to the next email. This is my code now

instance Job State EmailJob where
    job (State mvar) (SendToMailingList lId fAddress eSubject eContentType eContent) = do
        config    <- getConfig
        sessionId <- UUID.toString <$> UUID.nextRandom
        relations <- runReaderT (runDb $ selectList [MailingListAddressMailingList ==. mailingIdToKey lId] []) config
        let frAddress = Address fAddress ""
        currentTime <- liftIO getCurrentTime
        let mId = mailingIdToKey lId
        let sId = T.pack sessionId
        delivery <- runReaderT (runDb $ insert $ MailingListDelivery mId sId currentTime) config
        mapM (\x -> do
            ml <- runReaderT (runDb $ get (mailingListAddressAddress (entityVal x))) config
            addNewJob (SendEmail 
                      (mailAddressEmail (fromJust ml))
                      fAddress
                      eSubject
                      eContentType
                      eContent
                      (Just sId)
                      (Just lId))
            )
            relations
        return Success
    job _ (SendEmail toAddress fAddress eSubject eContentType eContent sId lId) = do
        let mId = case lId of 
                    Just listId -> Just $ mailingIdToKey listId
                    Nothing -> Nothing
        let content  = EmailContent eContentType eContent
        config    <- getConfig
        runReaderT (sendEmail 
                   (EmailMessage 
                   (Address toAddress "")
                   (Address fAddress  "")
                   eSubject 
                   content)
                   sId
                   mId)
                   config
        return Success

so in EmailJob I'm using mapM to create a new job for each email. SendEmail is the job which handles sending the email.
I'm on this email list and by using this method I got sent the same email twice. Note that duplicate emails don't exist because the mail API checks if an address exist before adding it.

I'm just wondering what is going wrong here. This is how I start the worker and the monitor. Should I skip the monitor to prevent this from happening? Each email job should not take 30 minutes so I'm wondering why it's not marking the job as complete

This is how I start the worker

main = do
    Dotenv.loadFile defaultConfig
    state <- newMVar 0

    host <- lookupEnv "REDIS_HOST"
    let connectInfo = defaultConnectInfo { connectHost = fromMaybe "localhost" host }

    let workerConfig = (defaultHworkerConfig "MailWorker" (State state)) 
                      { hwconfigDebug = True
                      , hwconfigRedisConnectInfo =  RedisConnectInfo connectInfo
                      , hwconfigTimeout = 1800
                      }

    sworker <- createWith workerConfig

    -- Start the worker
    forkIO (worker sworker)

    -- Start the worker monitor
    -- forkIO (monitor sworker)
    monitor sworker

[naglalakk]

I should note that addNewJob is my own function that I created to be able to add jobs

addNewJob :: EmailJob -> IO Integer
addNewJob job = do
    conf   <- getConfig
    jobStr <- createJob job
    jobAdded <- runReaderT (runCache $ R.lpush "hworker-jobs-MailWorker" [jobStr]) conf
    case jobAdded of
        Left _ -> return 0
        Right i -> return i

[naglalakk]

Sorry for bombing in messages here but just to update this briefly. I ran this locally with 3000 bs emails and included myself in the list. There is one thing I didn't notice before. On some of the jobs I get a libpq: failed (FATAL: remaining connection slots are reserved for non-replication superuser connections\n)

which is not connected to hworker but Postgres and Persistent. What is weird about this is that the queue is running a single job at a time so I don't know why it should be opening up to many connections. But because of this I'm trying to run it with a threadDelay of one second per job and it looks like it's more stable. I'm not getting any duplicate emails

[naglalakk]

I'm closing this issue since I cannot replicate this locally and it doesn't seem to be a hworker problem at all