Message acknowledgement with external subsystems

[beatrausch]

Hi,

I am looking for a robust way to acknowledge a message after being asynchronously processed by a subsystem e.g. some external library that is not under my control, thus I cannot pass the Msg and call Ack.

The current approach passes the reply string to the subsystem. When the subsystem responses I simply call Publish on the nats connection with either +ACK or -ACK. Unfortunately, ackAck and ackNak are private in the nats go client, thus it feels a bit hacky.

I am interested in your opinion about that solution and if there are plans to support such a use case in a simpler way e.g. by adding additional methods Ack, Nak, Term, etc. in the Jetstream interface.

Regards

[derekcollison]

They are all present on a received message, can you hold onto the message and when the subsystem responds just call msg.Ack() or msg.Nak()?

[beatrausch]

Unfortunately, I cannot hold the current message and wait until the subsystem responses. Due to the asynchronous, distributed design, it is not guaranteed that the acknowledgement is sent by the same goroutine which receives the Msg from NATS and to make more complicated it might not be the same process.

[derekcollison]

You could construct a nats.Msg on the fly with the reply subject and a subscription bound to the connection you want to use. Then you could use m.Ack() etc.

[kozlovic]

@derekcollison But that won't work since we check that the message is associated with a subscription, that the subscription is "JetStream" one. That's also how we get the connection. With a "bare" new message, there is no way Ack() can work.

[derekcollison]

Hmm, ok. Fair point. So I guess they will need to recreate all of the Ack calls for situations that just have a NATS connection and reply subject.

[beatrausch]

By creating a new nats.Msg from an Subscription, that comes from a JetStream, Ack calls on the newly created message works.

[tpihl]

How do you handle retries or multiple processing of the same info. I understand that the async subsystem is a black box but I don’t understand the solution design. Keeping the msg in a blocked goroutine until processing is done would make it possible to use acks the simple way (including extending and nak).

If you pick up the completion via some other means (webhook or similar), a normal nats request/reply would be a great way to “find” the right goroutine. Or if there is many many requests, get the nats msg from a map or something and then ack it.

Either way, mechanisms for idempotemcy and retry/fallback actions should also be considered (and that can be very hard to do unless “someone” holds the active msg.

[beatrausch]

How do you handle retries or multiple processing of the same info

Since at-least-once semantic in the data processing is fine, there is no special processing required. Furthermore, each message has in its payload a unique id that allows the downstream processing to identify messages that haven delivered multiple times into the subsystem.

I understand that the async subsystem is a black box but I don’t understand the solution design. Keeping the msg in a blocked goroutine until processing is done would make it possible to use acks the simple way (including extending and nak).

That’s absolutely true, but as I took over, the application design, was mostly done. May be that’s something that I could consider during future changes.