bug: handler's terminate task won't issue an event

[drev74]

Hi I'm not sure whether it's a bug or an expected behavior, but this needs to be fixed.

I have a simple handler:

// handler
func SearchByXid(_ context.Context, logger *zap.SugaredLogger, t *aj.Task) (any, error) {
  return "success", nil
  // return nil, aj.ErrTerminateTask
}

And an event listener logic:

// event listener
subscription, _ := nc.SubscribeSync(aj.EventsSubjectWildcard)
for {
	msg, err := subscription.NextMsg(5 * time.Second)
	if err != nil {
         logger.Error(route, zap.Error(err))			
	}

	event, kind, err := aj.ParseEventJSON(msg.Data)
	if err != nil {
	  fmt.Printf("Could not parse event: %v\n", err)
	}

	switch e := event.(type) {
	case aj.TaskStateChangeEvent:
          // process event
	default:
	   // process default
	}
}

For a happy 😄 case, the event listener receives the aj.TaskStateChangeEvent:

2025-05-02T13:50:35.477+0300    DEBUG   ajc     Enqueueing task into queue dev via CHORIA_AJ.Q.dev.2wXLac90JhuCwd0YBZgH88Vhy9o
[10:50:35] 2wXLac90JhuCwd0YBZgH88Vhy9o: queue: dev type: search:xid tries: 0 state: active
[10:50:35] 2wXLac90JhuCwd0YBZgH88Vhy9o: queue: dev type: search:xid tries: 1 state: terminated error: terminate task
2025-05-02T13:50:40.530+0300    ERROR   ui      callback        {"error": "nats: timeout"}

But all other handler responses, I'm getting NO events and just timing out:

2025-05-02T14:13:44.750+0300    DEBUG   ajc     Enqueueing task into queue dev via CHORIA_AJ.Q.dev.2wXOP7gITXsjZHfqQPWRZn7cpgs
2025-05-02T14:13:49.760+0300    ERROR   ui      callback        {"error": "nats: timeout"}

For a handler, which returns a success:

// handler
func SearchByXid(_ context.Context, logger *zap.SugaredLogger, t *aj.Task) (any, error) {
  return "success", nil
}

I'm getting an expected behavior:

2025-05-02T14:16:48.438+0300    DEBUG   ajc     Enqueueing task into queue dev via CHORIA_AJ.Q.dev.2wXOmEvZhwkvFVXTwRGKhwAPJBm
[11:16:48] 2wXOmEvZhwkvFVXTwRGKhwAPJBm: queue: dev type: search:xid tries: 0 state: active
[11:16:48] 2wXOmEvZhwkvFVXTwRGKhwAPJBm: queue: dev type: search:xid tries: 1 state: complete
2025-05-02T14:16:53.490+0300    ERROR   ui      callback        {"error": "nats: timeout"}

Bottom line: events on aj.ErrTerminateTask are getting lost 🙄

[ripienaar]

Appears to do the right thing, here:

asyncjobs/client.go

Lines 348 to 354 in bce47ac

	func (c *Client) handleTaskTerminated(ctx context.Context, t *Task, terr error) error {
	t.LastErr = terr.Error()
	t.LastTriedAt = nowPointer()
	t.State = TaskStateTerminated
	return c.saveOrDiscardTaskIfDesired(ctx, t)
	}

and here:

asyncjobs/client.go

Lines 324 to 333 in bce47ac

	func (c *Client) saveOrDiscardTaskIfDesired(ctx context.Context, t *Task) error {
	if !c.shouldDiscardTask(t) {
	return c.storage.SaveTaskState(ctx, t, true)
	}
	c.storage.PublishTaskStateChangeEvent(ctx, t)
	c.log.Debugf("Discarding task with state %s based on desired discards %q", t.State, c.opts.discard)
	return c.storage.DeleteTaskByID(t.ID)
	}

It should publish it unless its in your discard list?

[drev74]

It should publish it unless its in your discard list?

I don't use a discard list. Let me try to impl a test bench for this 🙄

[drev74]

@ripienaar

I couldn't reproduce the bug within the project, so I assume there's a problem in my environment 🙄
I added a test to cover this, how about merging it ? 🤔 #130

[ripienaar]

Do you perhaps exit your handler app soon after failing to handle? there are some small buffering happening in nats clients so perhaps it publishes but never gets a chance to flush to the server?

[drev74]

Do you perhaps exit your handler app soon after failing to handle?

I don't think so. I use DI and there may be some mess with it. I'll need to investigate regardless of this issue.
Closing this as non-reproducible