Repository: nats-io/nats-streaming-server (go)

Issue:
"duplicate durable registration" error after "subscribe request timeout"
I've been testing how my stan clients, written using nats.c, behave in the face of network problems introducing high packet loss (which, unfortunately, happens on rare occasions between two of our geographically dispersed sites). During my testing I've found that on some occasions, when resubscribing after a connection failure, the subscription request will fail with "subscribe request timeout" then, after retrying, will fail with "duplicate durable registration". All subsequent resubscribe attempts will fail with that message for the duration of the connection.

The reconnect logic my stan clients use is pretty straight forward and believe matches the recommendation:
* Set the ConnectionLostHandler on the stan connection
* The ConnectionLostHandler will:
  * Destroy all subscriptions
  * Destroy the connection
  * Attempt to reestablish the connection until it succeeds
  * Attempt to resubscribe each subscription until it succeeds unless the subscription fails due to NATS_CONNECTION_CLOSED in which case we stop everything and let the ConnectionLostHandler run again.

I'm simulating an unreliable network by using Linux's [NetEm](https://man7.org/linux/man-pages/man8/tc-netem.8.html). I've configured the network interface to rate limit to 150kbit and 40% packet loss (unrealistically high but appears to be quite good at hitting many possible failure cases when running for hours/days at a time). Most of the time the re-connection logic works correctly, except, of course, for the originally stated problem.

I haven't confirmed this but it appears as if the SubscriptionRequest message gets to the server but the SubscriptionResponse message doesn't get back to the client before the ConnectionWait period elapses causing a situation where the server thinks the client has a valid subscription but the client disagrees. This traps the client in a state where the subscription request can never be fulfilled until the connection stops and starts again.

Out of curiosity I tested the same scenario with nats-replicator. It too will get trapped in a "duplicate durable registration" error loop until the connection dies again.

Setting the ConnectionWait time to something higher (I've been using 10 seconds) significantly reduces the occurrence of the problem but does not entirely eliminate it.

The only work around I've been able to find is to simply destroy the connection entirely and reconnect when this error is encountered. However, that seems like a fairly drastic solution. Is there a better workaround for this?

Pull request:
[ADDED] ReplaceDurable option
When the subscription request for a durable subscription times out
or fail on the client side, but it was accepted in the server, then
if the application tries to restart the subscription request again
it will fail with a "duplicate durable subscription" error until
the connection is closed.

This new option allows the user to decide how the server should behave
when processing a duplicate durable subscription. If disabled, the
default, it behaves as described above, that is, it will reject
the second subscription request and return the "duplicate durable"
error.
If enabled, if the server detects that this is a duplicate, it will
close the active one and accept the new one. It is a suspend followed
by a resume.

From the client perspective, if this is done in the context of #1135,
then everything works well since the original subscription in the
client was actually not started due to subscription request failure.
However, if user try to create multiple duplicate durable subscriptions
for subscription requests (Subscribe() calls) that did not fail, then
their application will not be notified that the subscriptions that are
being replaced are replaced, but they will simply stop receiving messages
on those.

Resolves #1135

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>

Patch:
diff --git a/nats-streaming-server.go b/nats-streaming-server.go
index 8dde3b5..7920b79 100644
--- a/nats-streaming-server.go
+++ b/nats-streaming-server.go
@@ -52,6 +52,7 @@ Streaming Server Options:
           --encrypt <bool>               Specify if server should use encryption at rest
           --encryption_cipher <string>   Cipher to use for encryption. Currently support AES and CHAHA (ChaChaPoly). Defaults to AES
           --encryption_key <string>      Encryption Key. It is recommended to specify it through the NATS_STREAMING_ENCRYPTION_KEY environment variable instead
+          --replace_durable <bool>       Replace the existing durable subscription instead of reporting a duplicate durable error
 
 Streaming Server Clustering Options:
     --clustered <bool>                     Run the server in a clustered configuration (default: false)
diff --git a/server/conf.go b/server/conf.go
index 16c58aa..9925a06 100644
--- a/server/conf.go
+++ b/server/conf.go
@@ -199,6 +199,11 @@ func ProcessConfigFile(configFile string, opts *Options) error {
 				return err
 			}
 			opts.NKeySeedFile = v.(string)
+		case "replace_durable", "replace_durables", "replace_duplicate_durable", "replace_duplicate_durables":
+			if err := checkType(k, reflect.Bool, v); err != nil {
+				return err
+			}
+			opts.ReplaceDurable = v.(bool)
 		}
 	}
 	return nil
@@ -687,6 +692,7 @@ func ConfigureOptions(fs *flag.FlagSet, args []string, printVersion, printHelp,
 	fs.BoolVar(&sopts.Encrypt, "encrypt", false, "Specify if server should use encryption at rest")
 	fs.StringVar(&sopts.EncryptionCipher, "encryption_cipher", stores.CryptoCipherAutoSelect, "Encryption cipher. Supported are AES and CHACHA (default is AES)")
 	fs.StringVar(&encryptionKey, "encryption_key", "", "Encryption Key. It is recommended to specify it through the NATS_STREAMING_ENCRYPTION_KEY environment variable instead")
+	fs.BoolVar(&sopts.ReplaceDurable, "replace_durable", false, "Replace the existing durable subscription instead of reporting a duplicate durable error")
 
 	// First, we need to call NATS's ConfigureOptions() with above flag set.
 	// It will be augmented with NATS specific flags and call fs.Parse(args) for us.
diff --git a/server/server.go b/server/server.go
index 8758a6a..d75e8a4 100644
--- a/server/server.go
+++ b/server/server.go
@@ -1337,6 +1337,7 @@ type Options struct {
 	EncryptionKey      []byte        // Encryption key. The environment NATS_STREAMING_ENCRYPTION_KEY takes precedence and is the preferred way to provide the key.
 	Clustering         ClusteringOptions
 	NATSClientOpts     []nats.Option
+	ReplaceDurable     bool // If true, the subscription request for a durable subscription will replace the current durable instead of failing with duplicate durable error.
 }
 
 // Clone returns a deep copy of the Options object.
@@ -5109,6 +5110,11 @@ func (s *StanServer) processSubscriptionRequest(m *nats.Msg) {
 	// until we are done with this subscription. This will also stop
 	// the delete timer if one was set.
 	c, preventDelete, err := s.lookupOrCreateChannelPreventDelete(sr.Subject)
+	// For durable subscriptions, and if allowed, if this is going to be a
+	// duplicate, close the current durable and accept the new one.
+	if err == nil && sr.DurableName != "" && sr.QGroup == "" && s.opts.ReplaceDurable {
+		err = s.closeDurableIfDuplicate(c, sr)
+	}
 	if err == nil {
 		// If clustered, thread operations through Raft.
 		if s.isClustered {
@@ -5173,6 +5179,47 @@ func (s *StanServer) processSubscriptionRequest(m *nats.Msg) {
 	s.subStartCh <- &subStartInfo{c: c, sub: sub, qs: qs, isDurable: sub.IsDurable}
 }
 
+// This will close (and replicate the close operation if running in cluster mode)
+// the current durable subscription matching this subscription request information.
+// This should be invoked only if ReplaceDurable option is enabled.
+// It is used in case users want to be able to "resend" a subscription request
+// if the original request failed, due to timeout for instance. It could be that
+// the server accepted the original, sent the response back but the client library
+// gave up on it due to its own timeout. In that case, trying to issue the same
+// subscription request would lead to a "duplicate durable" error and the only choice
+// would be to close the connection.
+func (s *StanServer) closeDurableIfDuplicate(c *channel, sr *pb.SubscriptionRequest) error {
+	var duplicate bool
+	var ackInbox string
+	ss := c.ss
+	ss.RLock()
+	sub := ss.durables[durableKey(sr)]
+	if sub != nil {
+		sub.RLock()
+		duplicate = sub.ClientID != ""
+		ackInbox = sub.AckInbox
+		sub.RUnlock()
+	}
+	ss.RUnlock()
+	if !duplicate {
+		return nil
+	}
+	creq := &pb.UnsubscribeRequest{
+		ClientID: sr.ClientID,
+		Subject:  sr.Subject,
+		Inbox:    ackInbox,
+	}
+	var err error
+	if s.isClustered {
+		err = s.replicateCloseSubscription(creq)
+	} else {
+		s.closeMu.Lock()
+		err = s.unsubscribe(creq, true)
+		s.closeMu.Unlock()
+	}
+	return err
+}
+
 type subStateTraceCtx struct {
 	clientID      string
 	isRemove      bool


Test patch:
diff --git a/server/clustering_test.go b/server/clustering_test.go
index 287b585..95963c1 100644
--- a/server/clustering_test.go
+++ b/server/clustering_test.go
@@ -8031,3 +8031,53 @@ func TestClusteringAddRemoveClusterNodesWithBootstrap(t *testing.T) {
 		}
 	}
 }
+
+func TestClusteringDurableReplaced(t *testing.T) {
+	cleanupDatastore(t)
+	defer cleanupDatastore(t)
+	cleanupRaftLog(t)
+	defer cleanupRaftLog(t)
+
+	// For this test, use a central NATS server.
+	ns := natsdTest.RunDefaultServer()
+	defer ns.Shutdown()
+
+	// Configure first server
+	s1sOpts := getTestDefaultOptsForClustering("a", true)
+	s1sOpts.ReplaceDurable = true
+	s1 := runServerWithOpts(t, s1sOpts, nil)
+	defer s1.Shutdown()
+
+	// Wait for it to bootstrap
+	getLeader(t, 10*time.Second, s1)
+
+	// Configure second server.
+	s2sOpts := getTestDefaultOptsForClustering("b", false)
+	s2sOpts.ReplaceDurable = true
+	s2 := runServerWithOpts(t, s2sOpts, nil)
+	defer s2.Shutdown()
+
+	// Configure third server.
+	s3sOpts := getTestDefaultOptsForClustering("c", false)
+	s3sOpts.ReplaceDurable = true
+	s3 := runServerWithOpts(t, s3sOpts, nil)
+	defer s3.Shutdown()
+
+	testDurableReplaced(t, s1)
+
+	s1.Shutdown()
+	newLeader := getLeader(t, 2*time.Second, s2, s3)
+	c, err := newLeader.lookupOrCreateChannel("foo")
+	if err != nil {
+		t.Fatalf("Error looking up channel: %v", err)
+	}
+	if subs := c.ss.getAllSubs(); len(subs) != 0 {
+		t.Fatalf("Expected 0 sub, got %v", len(subs))
+	}
+	c.ss.RLock()
+	lenDur := len(c.ss.durables)
+	c.ss.RUnlock()
+	if lenDur != 1 {
+		t.Fatalf("Expected 1 durable, got %v", lenDur)
+	}
+}
diff --git a/server/conf_test.go b/server/conf_test.go
index 8636191..4a25f92 100644
--- a/server/conf_test.go
+++ b/server/conf_test.go
@@ -440,6 +440,7 @@ func TestParseWrongTypes(t *testing.T) {
 	expectFailureFor(t, "ft_group: 123", wrongTypeErr)
 	expectFailureFor(t, "partitioning: 123", wrongTypeErr)
 	expectFailureFor(t, "syslog_name: 123", wrongTypeErr)
+	expectFailureFor(t, "replace_durable: 123", wrongTypeErr)
 	expectFailureFor(t, "store_limits:{max_channels:false}", wrongTypeErr)
 	expectFailureFor(t, "store_limits:{max_msgs:false}", wrongTypeErr)
 	expectFailureFor(t, "store_limits:{max_bytes:false}", wrongTypeErr)
diff --git a/server/partitions_test.go b/server/partitions_test.go
index 278bea8..a93398e 100644
--- a/server/partitions_test.go
+++ b/server/partitions_test.go
@@ -1027,3 +1027,36 @@ func TestPartitionsCleanInvalidConns(t *testing.T) {
 		t.Fatalf("Should not be more than %v, got %v", maxKnownInvalidConns, mlen)
 	}
 }
+
+func TestPartitionsDurableReplaced(t *testing.T) {
+	setPartitionsVarsForTest()
+	defer resetDefaultPartitionsVars()
+
+	clientCheckTimeout = 150 * time.Millisecond
+	defer func() { clientCheckTimeout = defaultClientCheckTimeout }()
+
+	// For this test, create a single NATS server to which both servers connect to.
+	ns := natsdTest.RunDefaultServer()
+	defer ns.Shutdown()
+
+	fooSubj := "foo"
+	barSubj := "bar"
+
+	opts1 := GetDefaultOptions()
+	opts1.NATSServerURL = "nats://127.0.0.1:4222"
+	opts1.Partitioning = true
+	opts1.ReplaceDurable = true
+	opts1.StoreLimits.AddPerChannel(fooSubj, &stores.ChannelLimits{})
+	s1 := runServerWithOpts(t, opts1, nil)
+	defer s1.Shutdown()
+
+	opts2 := GetDefaultOptions()
+	opts2.NATSServerURL = "nats://127.0.0.1:4222"
+	opts2.Partitioning = true
+	opts2.ReplaceDurable = true
+	opts2.StoreLimits.AddPerChannel(barSubj, &stores.ChannelLimits{})
+	s2 := runServerWithOpts(t, opts2, nil)
+	defer s2.Shutdown()
+
+	testDurableReplaced(t, s1)
+}
diff --git a/server/server_durable_test.go b/server/server_durable_test.go
index f1b6018..278b298 100644
--- a/server/server_durable_test.go
+++ b/server/server_durable_test.go
@@ -16,6 +16,7 @@ package server
 import (
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -918,3 +919,53 @@ func TestPersistentStoreDurableClosedStatusOnRestart(t *testing.T) {
 	// Restart one last time
 	dur = restartDurable()
 }
+
+func TestDurableReplaced(t *testing.T) {
+	opts := GetDefaultOptions()
+	opts.ReplaceDurable = true
+	s := runServerWithOpts(t, opts, nil)
+	defer s.Shutdown()
+
+	testDurableReplaced(t, s)
+}
+
+func testDurableReplaced(t *testing.T, s *StanServer) {
+	sc := NewDefaultConnection(t)
+	defer sc.Close()
+
+	count := int32(0)
+	cb := func(_ *stan.Msg) {
+		atomic.AddInt32(&count, 1)
+	}
+	dur1, err := sc.Subscribe("foo", cb, stan.DurableName("dur"))
+	if err != nil {
+		t.Fatalf("Unable to start durable: %v", err)
+	}
+
+	dur2, err := sc.Subscribe("foo", cb, stan.DurableName("dur"))
+	if err != nil {
+		t.Fatalf("Unable to start durable: %v", err)
+	}
+
+	sc.Publish("foo", []byte("msg"))
+	time.Sleep(150 * time.Millisecond)
+
+	if c := atomic.LoadInt32(&count); c != 1 {
+		t.Fatalf("Received %v messages!", c)
+	}
+
+	c, err := s.lookupOrCreateChannel("foo")
+	if err != nil {
+		t.Fatalf("Error on lookup: %v", err)
+	}
+	if subs := c.ss.getAllSubs(); len(subs) > 1 {
+		t.Fatalf("Should have only 1 sub, got %v", len(subs))
+	}
+
+	if err := dur1.Close(); err == nil {
+		t.Fatal("Expected error on close of dur1, but did not get one")
+	}
+	if err := dur2.Close(); err != nil {
+		t.Fatalf("Error on close: %v", err)
+	}
+}