[IMPROVEMENT] added client retry for jetstream async publish old API #1695

pranavmehta94 · 2024-08-08T08:56:50Z

This PR contains following:

Added retry logic for jestream async publish using old API
retries are controlled by retry wait and retry attempts publish options
no responder error is retried
pub ack future object is re-used in retries

This PR fixes following issues:

PublishAsync Retry Mechanism Not Working as Expected #1678

piotrpio

Thanks for the contribution! I have a few comments.

piotrpio · 2024-08-09T10:21:37Z

js.go

+			dch := js.dch
+			js.dch = nil
+			// Defer here so error is processed and can be checked.
+			defer close(dch)


This defer does not work as expected since it'll be executed at the end of closeDch, not the caller. While it would be good to extract this func since it's repeated twice, I do think that we should make sure that close(dch) is executed at the very end to first send signals on the pubAckFuture and only later to PublishAsyncComplete()

piotrpio · 2024-08-09T10:25:01Z

js.go

@@ -894,7 +933,10 @@ func (js *js) PublishAsync(subj string, data []byte, opts ...PubOpt) (PubAckFutu
 const defaultStallWait = 200 * time.Millisecond

 func (js *js) PublishMsgAsync(m *Msg, opts ...PubOpt) (PubAckFuture, error) {
-	var o pubOpts
+	var o = pubOpts{
+		rwait: DefaultPubRetryWait,


Since this is an old API many people rely on, I would be hesitant to make the retries default, but rather treat them as opt-in and clearly state it in the doc. I know that this makes it work differently than the new API, but I don't think we should be changing the default behavior in this case.

piotrpio · 2024-08-09T10:28:09Z

js.go

+			js.dch = nil
+			// Defer here so error is processed and can be checked.
+			defer close(dch)
+		}
 	}

 	doErr := func(err error) {


We should clear the reply subject before passing the message in error handler. If user wants to retry manually inside the error handler, having paf.msg.Reply set will cause the retry attempt to fail with nats: reply subject should be empty

piotrpio · 2024-08-09T10:35:47Z

object.go

@@ -398,7 +398,7 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 		return nil
 	}

-	m, h := NewMsg(chunkSubj), sha256.New()


Why is this change necessary?

Earlier, reply field in msg was set to empty string on PublishMsgAsync exit using defer. After adding retry logic, go test detected race between setting reply in async handler handleAsyncReply and emptying it in PublishMsgAsync. So, the defer call which emptied the reply inside PublishMsgAsync was removed. The code was re-using the msg object inside the loop which called PublishMsgAsync and the test failed complaining about nats reply being not empty. Hence, the new msg object is being created in every iteration of the loop now

This is no good I think. We should not leave Msg.Reply set with the generated subject as it makes it impossible to re-publish by the user (as in the object store now). By the way, having Put() in object store recreate the message with each iteration means we have increased allocations (especially for large payloads / small chunks).

I may have an idea of how to tackle this problem - we could stop modifying the message passed by the user (which would prevent race) and utilize the low level nc.publish(). Here's my initial attempt from your branch: https://gist.github.com/piotrpio/796e5203e2d5be2cabda771e9ec51d54

Possibly this can be improved, but it seems to work.

// PublishMsgAsync publishes a Msg to JetStream and returns a PubAckFuture. // The message should not be changed until the PubAckFuture has been processed.

From the above comment, the way Put() uses PublishMsgAsync is incorrect since it seems to modify msg object without waiting on PubAckFuture.

Ok but even so, with your changes the message will never be restored to its original state right? Even after the PubAckFuture has been processed. If we need to fix it anyway we can also improve by not modifying the message at all I think.

I understand that a caller would assume that message would not be modified in anyway after PublishMsgAsync is called and PubAckFuture is processed. This is the reason why Reply was unset earlier. I will update the PR by applying your patch

But it was unset before your changes: https://github.com/nats-io/nats.go/pull/1695/files#diff-176a2342c55218b70f165bed6a7731dc83d901f9513382f3c98bbd3d08ce0678L941

After PubAckFuture is processed, the message can be re-used by the user and thus should be in its original state:
/ The message should not be changed *until* the PubAckFuture has been processed.

Again, your changes are good in general and we should definitely process those but we also need to be very careful when changing any behavior in the old JetStream API.

Yes I understand it was unset before these changes as well

@piotrpio I have updated the PR by applying your patch with one small change. Please review the changes

piotrpio · 2024-08-09T10:38:28Z

test/js_test.go

We should really have tests for async retries. For reference, see tests in the new API: https://github.com/nats-io/nats.go/blob/main/jetstream/test/publish_test.go#L1402

pranavmehta94 · 2024-08-09T17:50:38Z

@piotrpio I incorporated the review comments but the CI pipeline now fails in an unrelated test. When I run the tests locally, all the tests succeed

piotrpio

LGTM, thanks for the contribution!

…m API (#1695)

pranavmehta94 changed the title ~~added client retry for jetstream async publish old API~~ [IMPROVEMENT] added client retry for jetstream async publish old API Aug 8, 2024

added client retry for jetstream async publish old API

6bec9c3

pranavmehta94 force-pushed the async_publish_retry branch 2 times, most recently from e038a42 to 6753a97 Compare August 8, 2024 12:49

pranavmehta94 marked this pull request as draft August 8, 2024 13:43

fixed publish async test

9081c04

pranavmehta94 force-pushed the async_publish_retry branch from 6753a97 to 9081c04 Compare August 8, 2024 14:53

pranavmehta94 marked this pull request as ready for review August 8, 2024 15:04

piotrpio requested changes Aug 9, 2024

View reviewed changes

incorporated review comments and added tests for async retries

624cab3

pranavmehta94 force-pushed the async_publish_retry branch from 748e304 to 9fcc865 Compare August 10, 2024 01:59

added validation for retry attempts option in async publish

59f8525

pranavmehta94 force-pushed the async_publish_retry branch from 9fcc865 to 59f8525 Compare August 10, 2024 02:03

pranavmehta94 added 2 commits August 12, 2024 19:46

storing reply in PubAckFuture for re-use during retries

b55b7d0

resolving merge conflicts with main branch

0b4f4a7

piotrpio approved these changes Aug 14, 2024

View reviewed changes

piotrpio merged commit 6f181d3 into nats-io:main Aug 14, 2024
1 check passed

piotrpio pushed a commit that referenced this pull request Aug 15, 2024

[IMPROVEMENT] Added client retry for async publish in legacy JetStrea…

dffc184

…m API (#1695)

piotrpio pushed a commit that referenced this pull request Aug 15, 2024

[IMPROVEMENT] Added client retry for async publish in legacy JetStrea…

3cf526c

…m API (#1695)

piotrpio pushed a commit that referenced this pull request Dec 13, 2024

[IMPROVEMENT] Added client retry for async publish in legacy JetStrea…

575c4b4

…m API (#1695)

piotrpio pushed a commit that referenced this pull request Dec 13, 2024

[IMPROVEMENT] Added client retry for async publish in legacy JetStrea…

03f3421

…m API (#1695)

piotrpio mentioned this pull request Dec 17, 2024

Release v1.38.0 #1754

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[IMPROVEMENT] added client retry for jetstream async publish old API #1695

[IMPROVEMENT] added client retry for jetstream async publish old API #1695

pranavmehta94 commented Aug 8, 2024

piotrpio left a comment

piotrpio Aug 9, 2024

piotrpio Aug 9, 2024

piotrpio Aug 9, 2024

piotrpio Aug 9, 2024

pranavmehta94 Aug 9, 2024

piotrpio Aug 12, 2024

pranavmehta94 Aug 12, 2024

piotrpio Aug 12, 2024 •

edited

Loading

pranavmehta94 Aug 12, 2024 •

edited

Loading

piotrpio Aug 12, 2024

pranavmehta94 Aug 12, 2024

pranavmehta94 Aug 13, 2024 •

edited

Loading

piotrpio Aug 9, 2024

pranavmehta94 commented Aug 9, 2024

piotrpio left a comment

[IMPROVEMENT] added client retry for jetstream async publish old API #1695

[IMPROVEMENT] added client retry for jetstream async publish old API #1695

Conversation

pranavmehta94 commented Aug 8, 2024

piotrpio left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

piotrpio Aug 12, 2024 • edited Loading

Choose a reason for hiding this comment

pranavmehta94 Aug 12, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pranavmehta94 Aug 13, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pranavmehta94 commented Aug 9, 2024

piotrpio left a comment

Choose a reason for hiding this comment

piotrpio Aug 12, 2024 •

edited

Loading

pranavmehta94 Aug 12, 2024 •

edited

Loading

pranavmehta94 Aug 13, 2024 •

edited

Loading