Repository: nats-io/nats-server
Pull request #3352 (2022-08-09 21:13:01)

adding source/mirror unit tests
Signed-off-by: Matthias Hanel <mh@synadia.com>

 - [ ] Link to issue, e.g. `Resolves #NNN`
 - [ ] Documentation added (if applicable)
 - [ ] Tests added
 - [ ] Branch rebased on top of current main (`git pull --rebase origin main`)
 - [ ] Changes squashed to a single commit (described [here](http://gitready.com/advanced/2009/02/10/squashing-commits-with-rebase.html))
 - [ ] Build is green in Travis CI
 - [ ] You have certified that the contribution is your original work and that you license the work to the project under the [Apache 2 license](https://github.com/nats-io/nats-server/blob/main/LICENSE)

Resolves #

### Changes proposed in this pull request:

 -
 -
 -

/cc @nats-io/core



Patch:
diff --git a/server/jetstream.go b/server/jetstream.go
index 977b47fa..24580346 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -1748,7 +1748,7 @@ func (jsa *jsAccount) updateUsage(tierName string, storeType StorageType, delta
 	}
 }
 
-const usageTick = 1500 * time.Millisecond
+var usageTick = 1500 * time.Millisecond
 
 func (jsa *jsAccount) sendClusterUsageUpdateTimer() {
 	jsa.usageMu.Lock()
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index b9530142..3992618e 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -1478,6 +1478,7 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, rm *recoveryRemovals) (b
 				if isRecovering {
 					js.setStreamAssignmentRecovering(sa)
 				}
+
 				js.processUpdateStreamAssignment(sa)
 			default:
 				panic("JetStream Cluster Unknown meta entry op type")
@@ -1647,7 +1648,9 @@ func currentPeerCount(ci *ClusterInfo, peerSet []string, leaderId string) (curre
 func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnapshot bool) {
 	s, cc := js.server(), js.cluster
 	defer s.grWG.Done()
-
+	if mset != nil {
+		defer mset.monitorWg.Done()
+	}
 	js.mu.RLock()
 	n := sa.Group.node
 	meta := cc.meta
@@ -2008,6 +2011,8 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 			// If we were successful lookup up our stream now.
 			if err == nil {
 				if mset, err = acc.lookupStream(sa.Config.Name); mset != nil {
+					mset.monitorWg.Add(1)
+					defer mset.monitorWg.Done()
 					mset.setStreamAssignment(sa)
 					// Make sure to update our updateC which would have been nil.
 					uch = mset.updateC()
@@ -2733,7 +2738,11 @@ func (js *jetStream) processUpdateStreamAssignment(sa *streamAssignment) {
 				node.StepDown(sa.Group.Preferred)
 			}
 			node.ProposeRemovePeer(ourID)
+			// shut down monitor by shutting down raft
+			node.Delete()
 		}
+		// wait for monitor to be shut down
+		mset.monitorWg.Wait()
 		mset.stop(true, false)
 	}
 }
@@ -2763,6 +2772,7 @@ func (js *jetStream) processClusterUpdateStream(acc *Account, osa, sa *streamAss
 				mset.setLeader(false)
 				js.createRaftGroup(acc.GetName(), rg, storage)
 			}
+			mset.monitorWg.Add(1)
 			// Start monitoring..
 			s.startGoRoutine(func() { js.monitorStream(mset, sa, needsNode) })
 		} else if numReplicas == 1 && alreadyRunning {
@@ -2938,6 +2948,9 @@ func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignme
 	// Start our monitoring routine.
 	if rg.node != nil {
 		if !alreadyRunning {
+			if mset != nil {
+				mset.monitorWg.Add(1)
+			}
 			s.startGoRoutine(func() { js.monitorStream(mset, sa, false) })
 		}
 	} else {
@@ -3085,6 +3098,12 @@ func (js *jetStream) processClusterDeleteStream(sa *streamAssignment, isMember,
 	// Go ahead and delete the stream if we have it and the account here.
 	if acc, _ = s.LookupAccount(sa.Client.serviceAccount()); acc != nil {
 		if mset, _ := acc.lookupStream(sa.Config.Name); mset != nil {
+			// shut down monitor by shutting down raft
+			if n := mset.raftNode(); n != nil {
+				n.Delete()
+			}
+			// wait for monitor to be shut down
+			mset.monitorWg.Wait()
 			err = mset.stop(true, wasLeader)
 			stopped = true
 		}
diff --git a/server/stream.go b/server/stream.go
index 159dbe84..af36aaa3 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -237,6 +237,8 @@ type stream struct {
 	// Direct get subscription.
 	directSub *subscription
 	lastBySub *subscription
+
+	monitorWg sync.WaitGroup
 }
 
 type sourceInfo struct {


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index d97a78cc..1368dc19 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -5436,7 +5436,7 @@ func TestJetStreamClusterSourcesFilterSubjectUpdate(t *testing.T) {
 
 }
 
-func TestJetStreamClusterMirrorAndSourcesClusterRestart(t *testing.T) {
+func TestJetStreamClusterSourcesUpdateOriginError(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "MSR", 5)
 	defer c.shutdown()
 
@@ -5444,25 +5444,6 @@ func TestJetStreamClusterMirrorAndSourcesClusterRestart(t *testing.T) {
 	nc, js := jsClientConnect(t, c.randomServer())
 	defer nc.Close()
 
-	// Origin
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:     "TEST",
-		Subjects: []string{"foo", "bar", "baz.*"},
-		Replicas: 2,
-	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	// Create Mirror now.
-	_, err = js.AddStream(&nats.StreamConfig{
-		Name:     "M",
-		Mirror:   &nats.StreamSource{Name: "TEST"},
-		Replicas: 2,
-	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-
 	sendBatch := func(subject string, n int) {
 		t.Helper()
 		// Send a batch to a given subject.
@@ -5473,28 +5454,60 @@ func TestJetStreamClusterMirrorAndSourcesClusterRestart(t *testing.T) {
 		}
 	}
 
-	checkSync := func() {
+	checkSync := func(msgsTest, msgsM uint64) {
 		t.Helper()
 		checkFor(t, 10*time.Second, 500*time.Millisecond, func() error {
-			tsi, err := js.StreamInfo("TEST")
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-			msi, err := js.StreamInfo("M")
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-			if tsi.State.Msgs != msi.State.Msgs {
-				return fmt.Errorf("Total messages not the same: TEST %d vs M %d", tsi.State.Msgs, msi.State.Msgs)
+			if tsi, err := js.StreamInfo("TEST"); err != nil {
+				return err
+			} else if msi, err := js.StreamInfo("M"); err != nil {
+				return err
+			} else if tsi.State.Msgs != msgsTest {
+				return fmt.Errorf("received %d msgs from TEST, expected %d", tsi.State.Msgs, msgsTest)
+			} else if msi.State.Msgs != msgsM {
+				return fmt.Errorf("received %d msgs from TEST, expected %d", msi.State.Msgs, msgsM)
 			}
 			return nil
 		})
 	}
 
+	// Origin
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 2,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "M",
+		Sources:  []*nats.StreamSource{{Name: "TEST", FilterSubject: "foo"}},
+		Replicas: 2,
+	})
+
+	require_NoError(t, err)
+
 	// Send 100 msgs.
 	sendBatch("foo", 100)
-	checkSync()
+	checkSync(100, 100)
+
+	// update makes source invalid
+	_, err = js.UpdateStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"bar"},
+		Replicas: 2,
+	})
+	require_NoError(t, err)
+
+	// TODO check for downstream error propagation
 
+	_, err = js.Publish("foo", nil)
+	require_Error(t, err)
+
+	sendBatch("bar", 100)
+	// The source stream remains at 100 msgs as it still uses foo as it's filter
+	checkSync(200, 100)
+
+	nc.Close()
 	c.stopAll()
 	c.restartAll()
 	c.waitOnStreamLeader("$G", "TEST")
@@ -5503,8 +5516,118 @@ func TestJetStreamClusterMirrorAndSourcesClusterRestart(t *testing.T) {
 	nc, js = jsClientConnect(t, c.randomServer())
 	defer nc.Close()
 
+	checkSync(200, 100)
+
+	_, err = js.Publish("foo", nil)
+	require_Error(t, err)
+	require_Equal(t, err.Error(), "nats: no response from stream")
+
 	sendBatch("bar", 100)
-	checkSync()
+	// The source stream remains at 100 msgs as it still uses foo as it's filter
+	checkSync(300, 100)
+}
+
+func TestJetStreamClusterMirrorAndSourcesClusterRestart(t *testing.T) {
+	test := func(t *testing.T, mirror bool, filter bool) {
+		c := createJetStreamClusterExplicit(t, "MSR", 5)
+		defer c.shutdown()
+
+		// Client for API requests.
+		nc, js := jsClientConnect(t, c.randomServer())
+		defer nc.Close()
+
+		// Origin
+		_, err := js.AddStream(&nats.StreamConfig{
+			Name:     "TEST",
+			Subjects: []string{"foo", "bar", "baz.*"},
+			Replicas: 2,
+		})
+		require_NoError(t, err)
+
+		filterSubj := _EMPTY_
+		if filter {
+			filterSubj = "foo"
+		}
+
+		// Create Mirror/Source now.
+		if mirror {
+			_, err = js.AddStream(&nats.StreamConfig{
+				Name:     "M",
+				Mirror:   &nats.StreamSource{Name: "TEST", FilterSubject: filterSubj},
+				Replicas: 2,
+			})
+		} else {
+			_, err = js.AddStream(&nats.StreamConfig{
+				Name:     "M",
+				Sources:  []*nats.StreamSource{{Name: "TEST", FilterSubject: filterSubj}},
+				Replicas: 2,
+			})
+		}
+		require_NoError(t, err)
+
+		expectedMsgCount := uint64(0)
+
+		sendBatch := func(subject string, n int) {
+			t.Helper()
+			if subject == "foo" || !filter {
+				expectedMsgCount += uint64(n)
+			}
+			// Send a batch to a given subject.
+			for i := 0; i < n; i++ {
+				if _, err := js.Publish(subject, []byte("OK")); err != nil {
+					t.Fatalf("Unexpected publish error: %v", err)
+				}
+			}
+		}
+
+		checkSync := func(msgsTest, msgsM uint64) {
+			t.Helper()
+			checkFor(t, 10*time.Second, 500*time.Millisecond, func() error {
+				if tsi, err := js.StreamInfo("TEST"); err != nil {
+					return err
+				} else if msi, err := js.StreamInfo("M"); err != nil {
+					return err
+				} else if tsi.State.Msgs != msgsTest {
+					return fmt.Errorf("received %d msgs from TEST, expected %d", tsi.State.Msgs, msgsTest)
+				} else if msi.State.Msgs != msgsM {
+					return fmt.Errorf("received %d msgs from M, expected %d", msi.State.Msgs, msgsM)
+				}
+				return nil
+			})
+		}
+
+		sendBatch("foo", 100)
+		checkSync(100, expectedMsgCount)
+		sendBatch("bar", 100)
+		checkSync(200, expectedMsgCount)
+
+		nc.Close()
+		c.stopAll()
+		c.restartAll()
+		c.waitOnStreamLeader("$G", "TEST")
+		c.waitOnStreamLeader("$G", "M")
+
+		nc, js = jsClientConnect(t, c.randomServer())
+		defer nc.Close()
+
+		checkSync(200, expectedMsgCount)
+		sendBatch("foo", 100)
+		checkSync(300, expectedMsgCount)
+		sendBatch("bar", 100)
+		checkSync(400, expectedMsgCount)
+	}
+	t.Run("mirror-filter", func(t *testing.T) {
+		test(t, true, true)
+	})
+	t.Run("mirror-nofilter", func(t *testing.T) {
+		test(t, true, false)
+	})
+	t.Run("source-filter", func(t *testing.T) {
+		test(t, false, true)
+	})
+	t.Run("source-nofilter", func(t *testing.T) {
+		test(t, false, false)
+	})
 }
 
 func TestJetStreamClusterSourceFilterSubjectUpdateFail(t *testing.T) {
diff --git a/server/jetstream_super_cluster_test.go b/server/jetstream_super_cluster_test.go
index 70715e45..1e3fd644 100644
--- a/server/jetstream_super_cluster_test.go
+++ b/server/jetstream_super_cluster_test.go
@@ -2713,6 +2713,12 @@ func TestJetStreamSuperClusterTagInducedMoveCancel(t *testing.T) {
 }
 
 func TestJetStreamSuperClusterMoveCancel(t *testing.T) {
+	usageTickOld := usageTick
+	usageTick = 250 * time.Millisecond
+	defer func() {
+		usageTick = usageTickOld
+	}()
+
 	server := map[string]struct{}{}
 	sc := createJetStreamSuperClusterWithTemplateAndModHook(t, jsClusterTempl, 4, 2,
 		func(serverName, clusterName, storeDir, conf string) string {
@@ -2821,6 +2827,10 @@ func TestJetStreamSuperClusterMoveCancel(t *testing.T) {
 	require_NoError(t, err)
 	defer ncsys.Close()
 
+	time.Sleep(2 * usageTick)
+	aiBefore, err := js.AccountInfo()
+	require_NoError(t, err)
+
 	for _, moveFromSrv := range streamPeerSrv {
 		moveReq, err := json.Marshal(&JSApiMetaServerStreamMoveRequest{Server: moveFromSrv, Tags: []string{emptySrv}})
 		require_NoError(t, err)
@@ -2846,6 +2856,15 @@ func TestJetStreamSuperClusterMoveCancel(t *testing.T) {
 			checkFor(t, 20*time.Second, 100*time.Millisecond, func() error { return checkSrvInvariant(s, expectedPeers) })
 		}
 		checkFor(t, 10*time.Second, 100*time.Millisecond, func() error { return serverEmpty(emptySrv) })
+		checkFor(t, 3*usageTick, 100*time.Millisecond, func() error {
+			if aiAfter, err := js.AccountInfo(); err != nil {
+				return err
+			} else if aiAfter.Store != aiBefore.Store {
+				return fmt.Errorf("store before %d and after %d don't match", aiBefore.Store, aiAfter.Store)
+			} else {
+				return nil
+			}
+		})
 	}
 }