FR: Acknowledging jobs outside of worker function

[rauanmayemir]

Currently, the job is 'done' as long as worker.Work() returns.
I need a different approach similar to what AMQP supports where pulling a job is decoupled from acknowledging or not acknowledging it.

Ideally, some sort of AsyncWorker would help a lot, especially if it can be handled with channels. I.e something like:

consumeChan, err = riverClient.Consume(ctx, &river.ConsumeOpts{
  // I'm not sure this makes sense here in case of pg notify
  MaxPrefechCount: 10,
  // Simimar to `RescueStuckJobsAfter`, nack the stuck jobs after a timeout
  NackTimeout: time.Second * time.Duration(60)
})

go func() {
  for asyncJob := range consumeChan {
    processJob(asyncJob)
  }
}()


func processJob(asyncJob *river.AsyncJob[AsyncJobArgs])  {
    // autonomous `worker.Work` behaviour
    // process the job ()
    // ...

    if err != nil {
      _ = asyncJob.JobSnooze(err)
      // too many retries
      // _ = asyncJob.JobCancel(err)
    }

    err = asyncJob.JobComplete()
    if err != nil {
      log.Debug("failed to complete the job, channel is probably closed")
    }

}

[bgentry]

Hmm, I don’t think this is likely going to be something we would build within River itself. The entire model is predicated on jobs being worked by a worker’s Work method.

You may be able to experiment with some different approaches to try and bridge the gap within user space, but fundamentally the job will stay as executing until the Work function returns.

[rauanmayemir]

I'd like to give some context on my use case.

Our backend is in PHP and the app server is in go (the model is pretty popular these days: https://roadrunner.dev, https://frankenphp.dev).

The idea was to insert jobs in transactional scope from PHP side same as what river does with client SDKs for Python and Ruby (that part is trivial to solve).
On the go side, a plugin would integrate with river to read jobs from db, but instead of a simple Work() function call, there is a slightly more complicated infra with a priority queue where jobs must be sent into an internal PQ from which the jobs are further distributed into PHP worker processes. Upon completing or failing, jobs are then nacked or acked.

Unless we block inside Work(), what are our options here?

[travisby]

I've found a fanout approach to work for similar problems rauanmayemir. PHP creates JobA, go immediately "works" it by creating JobB, potentially in another queue.

Although it almost sounds like you want to run PHP for the Work() functions? If you take the above approach would os.Exec("php" work together?

[rauanmayemir]

Yeah, that’s what I’m doing, except it’s a process pool and I send to an async queue instead of running os.Exec.