Repository: nats-io/nats.go
Pull request #1391 (2023-09-07 14:41:17)

[FIXED] Sending publish async completion signal after reconnect


Patch:
diff --git a/jetstream/publish.go b/jetstream/publish.go
index 1215687..30371cd 100644
--- a/jetstream/publish.go
+++ b/jetstream/publish.go
@@ -402,6 +402,10 @@ func (js *jetStream) resetPendingAcksOnReconnect() {
 			paf.err = nats.ErrDisconnected
 		}
 		js.publisher.acks = nil
+		if js.publisher.doneCh != nil {
+			close(js.publisher.doneCh)
+			js.publisher.doneCh = nil
+		}
 		js.publisher.Unlock()
 	}
 }
diff --git a/js.go b/js.go
index 05637d5..a8deb98 100644
--- a/js.go
+++ b/js.go
@@ -698,6 +698,10 @@ func (js *js) resetPendingAcksOnReconnect() {
 			paf.err = ErrDisconnected
 		}
 		js.pafs = nil
+		if js.dch != nil {
+			close(js.dch)
+			js.dch = nil
+		}
 		js.mu.Unlock()
 	}
 }
diff --git a/nats.go b/nats.go
index 8b9fd88..ba262b4 100644
--- a/nats.go
+++ b/nats.go
@@ -946,6 +946,7 @@ func ReconnectWait(t time.Duration) Option {
 }
 
 // MaxReconnects is an Option to set the maximum number of reconnect attempts.
+// If negative, it will never stop trying to reconnect.
 // Defaults to 60.
 func MaxReconnects(max int) Option {
 	return func(o *Options) error {
@@ -2500,6 +2501,9 @@ func (nc *Conn) sendConnect() error {
 	// Construct the CONNECT protocol string
 	cProto, err := nc.connectProto()
 	if err != nil {
+		if !nc.initc && nc.Opts.AsyncErrorCB != nil {
+			nc.ach.push(func() { nc.Opts.AsyncErrorCB(nc, nil, err) })
+		}
 		return err
 	}
 


Test patch:
diff --git a/jetstream/test/publish_test.go b/jetstream/test/publish_test.go
index 80ddc71..ac7a34c 100644
--- a/jetstream/test/publish_test.go
+++ b/jetstream/test/publish_test.go
@@ -1233,6 +1233,43 @@ func TestPublishMsgAsyncWithPendingMsgs(t *testing.T) {
 			t.Fatalf("Expected error: %v; got: %v", jetstream.ErrTooManyStalledMsgs, err)
 		}
 	})
+
+	t.Run("with server restart", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		_, err = js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		go func() {
+			for i := 0; i < 50; i++ {
+				_, _ = js.PublishAsync("FOO.1", []byte("msg"))
+			}
+		}()
+		srv = restartBasicJSServer(t, srv)
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+
+		select {
+		case <-js.PublishAsyncComplete():
+		case <-time.After(10 * time.Second):
+			t.Fatalf("Did not receive completion signal")
+		}
+	})
 }
 
 func TestPublishAsyncResetPendingOnReconnect(t *testing.T) {
diff --git a/nats_test.go b/nats_test.go
index 69cfb28..b70a190 100644
--- a/nats_test.go
+++ b/nats_test.go
@@ -1575,6 +1575,38 @@ func TestUserCredentialsChainedFile(t *testing.T) {
 	}
 }
 
+func TestReconnectMissingCredentials(t *testing.T) {
+	ts := runTrustServer()
+	defer ts.Shutdown()
+
+	chainedFile := createTmpFile(t, []byte(chained))
+	defer os.Remove(chainedFile)
+
+	url := fmt.Sprintf("nats://127.0.0.1:%d", TEST_PORT)
+	errs := make(chan error, 1)
+	nc, err := Connect(url, UserCredentials(chainedFile), ErrorHandler(func(_ *Conn, _ *Subscription, err error) {
+		errs <- err
+	}))
+	if err != nil {
+		t.Fatalf("Expected to connect, got %v", err)
+	}
+	defer nc.Close()
+	os.Remove(chainedFile)
+	ts.Shutdown()
+
+	ts = runTrustServer()
+	defer ts.Shutdown()
+
+	select {
+	case err := <-errs:
+		if !strings.Contains(err.Error(), "no such file or directory") {
+			t.Fatalf("Expected error about missing creds file, got %q", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("Did not get error about missing creds file")
+	}
+}
+
 func TestUserJWTAndSeed(t *testing.T) {
 	if server.VERSION[0] == '1' {
 		t.Skip()