Release v2.10.8 cherry picks (#4887)

Includes:

- Check all filterSubjects of consumers when purging a stream (#4873)
- Move tokenizing of subject after early returns (#4880)
- Move helper subjectInfo() to the file where it's used (#4881)
- JSAPI internal routing and reporting and Source and Mirror setup (#4884)
- Do not load all blocks for NumPending when delivery is LastPerSubject. (#4885)
- Detect corrupt psim subjects during recovery of index.db (#4890)
- Add 'certs' option to TLS block for multi-certs support (#4889)

server/config_check_test.go

@@ -298,7 +298,7 @@ func TestConfigCheck(t *testing.T) {
 		  hello = "world"
 		}
 		`,
-			err:       errors.New(`error parsing tls config, unknown field ["hello"]`),
+			err:       errors.New(`error parsing tls config, unknown field "hello"`),
 			errorLine: 3,
 			errorPos:  5,
 		},
@@ -311,7 +311,7 @@ func TestConfigCheck(t *testing.T) {
 		  }
 		}
 		`,
-			err:       errors.New(`error parsing tls config, unknown field ["foo"]`),
+			err:       errors.New(`error parsing tls config, unknown field "foo"`),
 			errorLine: 4,
 			errorPos:  7,
 		},
@@ -326,7 +326,7 @@ func TestConfigCheck(t *testing.T) {
 		    preferences = []
 		}
 		`,
-			err:       errors.New(`error parsing tls config, unknown field ["preferences"]`),
+			err:       errors.New(`error parsing tls config, unknown field "preferences"`),
 			errorLine: 7,
 			errorPos:  7,
 		},
@@ -342,7 +342,7 @@ func TestConfigCheck(t *testing.T) {
 		    suites = []
 		}
 		`,
-			err:       errors.New(`error parsing tls config, unknown field ["suites"]`),
+			err:       errors.New(`error parsing tls config, unknown field "suites"`),
 			errorLine: 8,
 			errorPos:  7,
 		},
@@ -1832,6 +1832,93 @@ func TestConfigCheck(t *testing.T) {
 			errorLine: 4,
 			errorPos:  6,
 		},
+		{
+			name: "TLS multiple certs",
+			config: `
+				port: -1
+				tls {
+					certs: [
+					  { cert_file: "configs/certs/server.pem", key_file: "configs/certs/key.pem"},
+					  { cert_file: "configs/certs/cert.new.pem", key_file: "configs/certs/key.new.pem"},
+					]
+				}
+			`,
+			err: nil,
+		},
+		{
+			name: "TLS multiple certs, bad type",
+			config: `
+				port: -1
+				tls {
+					certs: [
+					  { cert_file: "configs/certs/server.pem", key_file: 123 },
+					  { cert_file: "configs/certs/cert.new.pem", key_file: "configs/certs/key.new.pem"},
+					]
+				}
+			`,
+			err:       fmt.Errorf("error parsing certificates config: unsupported type int64"),
+			errorLine: 5,
+			errorPos:  49,
+		},
+		{
+			name: "TLS multiple certs, missing key_file",
+			config: `
+				port: -1
+				tls {
+					certs: [
+					  { cert_file: "configs/certs/server.pem" }
+					  { cert_file: "configs/certs/cert.new.pem", key_file: "configs/certs/key.new.pem"}
+					]
+				}
+			`,
+			err:       fmt.Errorf("error parsing certificates config: both 'cert_file' and 'cert_key' options are required"),
+			errorLine: 5,
+			errorPos:  10,
+		},
+		{
+			name: "TLS multiple certs and single cert options at the same time",
+			config: `
+				port: -1
+				tls {
+					cert_file: "configs/certs/server.pem"
+					key_file: "configs/certs/key.pem"
+					certs: [
+					  { cert_file: "configs/certs/server.pem", key_file: "configs/certs/key.pem"},
+					  { cert_file: "configs/certs/cert.new.pem", key_file: "configs/certs/key.new.pem"},
+					]
+				}
+			`,
+			err:       fmt.Errorf("error parsing tls config, cannot combine 'cert_file' option with 'certs' option"),
+			errorLine: 3,
+			errorPos:  5,
+		},
+		{
+			name: "TLS multiple certs used but not configured, but cert_file configured",
+			config: `
+				port: -1
+				tls {
+					cert_file: "configs/certs/server.pem"
+					key_file: "configs/certs/key.pem"
+					certs: []
+				}
+			`,
+			err: nil,
+		},
+		{
+			name: "TLS multiple certs, missing bad path",
+			config: `
+				port: -1
+				tls {
+					certs: [
+					  { cert_file: "configs/certs/cert.new.pem", key_file: "configs/certs/key.new.pem"}
+					  { cert_file: "configs/certs/server.pem", key_file: "configs/certs/key.new.pom" }
+					]
+				}
+			`,
+			err:       fmt.Errorf("error parsing X509 certificate/key pair 2/2: open configs/certs/key.new.pom: no such file or directory"),
+			errorLine: 3,
+			errorPos:  5,
+		},
 	}
 
 	checkConfig := func(config string) error {

server/configs/reload/tls_multi_cert_1.conf

@@ -0,0 +1,13 @@
+# Simple TLS config file
+
+listen:   127.0.0.1:-1
+
+tls {
+  certs = [
+    {
+      cert_file: "../test/configs/certs/srva-cert.pem"
+      key_file:  "../test/configs/certs/srva-key.pem"
+    }
+  ]
+  timeout:    2
+}

server/configs/reload/tls_multi_cert_2.conf

@@ -0,0 +1,19 @@
+# Simple TLS config file
+
+listen:   127.0.0.1:-1
+
+tls {
+  certs = [
+    {
+      cert_file: "../test/configs/certs/srva-cert.pem"
+      key_file:  "../test/configs/certs/srva-key.pem"
+    },
+    {
+      cert_file: "../test/configs/certs/srvb-cert.pem"
+      key_file:  "../test/configs/certs/srvb-key.pem"
+    }
+  ]
+  ca_file: "../test/configs/certs/ca.pem"
+  verify:  true
+  timeout: 2
+}

server/configs/reload/tls_multi_cert_3.conf

@@ -0,0 +1,13 @@
+# Simple TLS config file
+
+listen:   127.0.0.1:-1
+
+tls {
+  certs = [
+    {
+      cert_file: "../test/configs/certs/srvb-cert.pem"
+      key_file:  "../test/configs/certs/srvb-key.pem"
+    }
+  ]
+  timeout: 2
+}

server/consumer.go

@@ -3350,7 +3350,7 @@ func (o *consumer) notifyDeliveryExceeded(sseq, dc uint64) {
 	o.sendAdvisory(o.deliveryExcEventT, j)
 }
 
-// Check to see if the candidate subject matches a filter if its present.
+// Check if the candidate subject matches a filter if its present.
 // Lock should be held.
 func (o *consumer) isFilteredMatch(subj string) bool {
 	// No filter is automatic match.
@@ -3374,6 +3374,25 @@ func (o *consumer) isFilteredMatch(subj string) bool {
 	return false
 }
 
+// Check if the candidate filter subject is equal to or a subset match
+// of one of the filter subjects.
+// Lock should be held.
+func (o *consumer) isEqualOrSubsetMatch(subj string) bool {
+	for _, filter := range o.subjf {
+		if !filter.hasWildcard && subj == filter.subject {
+			return true
+		}
+	}
+	tsa := [32]string{}
+	tts := tokenizeSubjectIntoSlice(tsa[:0], subj)
+	for _, filter := range o.subjf {
+		if isSubsetMatchTokenized(filter.tokenizedSubject, tts) {
+			return true
+		}
+	}
+	return false
+}
+
 var (
 	errMaxAckPending = errors.New("max ack pending reached")
 	errBadConsumer   = errors.New("consumer not valid")