Repository: nats-io/nats-server
Pull request #1737 (2020-11-23 05:02:24)

Cross-compile for FreeBSD/amd64 without cgo
(These commits should be preserved in merge, not squashed.)

The first commit fixes FreeBSD %cpu calculation, which was wrong by a factor of
2048, and provides an alternative implementation on amd64 which avoids the need
for cgo.  The framework setup is extensible to other architectures.

The second commit adds goreleaser targeting FreeBSD/amd64 as part of our usual
release artifact generation.

f088660d
   goreleaser target freebsd/amd64

80d11fa1
   FreeBSD fixes: %cpu scale; no-cgo for amd64

   It would be convenient if we did not need to use cgo to compile for FreeBSD
   for common architectures, allowing builds to be generated in Linux CI
   flows.

   The sysctl interface uses offsets which can vary by build architecture, so
   doing this in the general case without cgo is not realistic.  But we can
   front-load the C work to get the offsets for a given architecture, then use
   encoding/binary at run-time.

   While doing this, I saw that the existing FreeBSD code was mis-calculating
   `%cpu` by converting the `fixpt_t` scaled int straight to a double without
   dividing by the scaling factor, so we also fix for all other architectures
   by introducing a division by `FSCALE`.

   The offsets-emitting code is in `freebsd.txt`, with the filename chosen to
   keep the Go toolchain from picking it up and trying to compile.

   The result is unsafe-free cgo-free builds for FreeBSD/amd64.

/cc @nats-io/core


Patch:
diff --git a/.goreleaser.yml b/.goreleaser.yml
index 8ce80fe9..5632ae7c 100644
--- a/.goreleaser.yml
+++ b/.goreleaser.yml
@@ -22,6 +22,7 @@ builds:
   - darwin
   - linux
   - windows
+  - freebsd
   goarch:
   - amd64
   - arm
@@ -33,6 +34,12 @@ builds:
   ignore:
     - goos: darwin
       goarch: 386
+    - goos: freebsd
+      goarch: arm
+    - goos: freebsd
+      goarch: arm64
+    - goos: freebsd
+      goarch: 386
 
 nfpms:
   - file_name_template: '{{.ProjectName}}-{{.Tag}}-{{.Arch}}{{if .Arm}}{{.Arm}}{{end}}'
diff --git a/server/accounts.go b/server/accounts.go
index c5d379d3..9a68c032 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -1344,10 +1344,26 @@ func (a *Account) AddServiceImportWithClaim(destination *Account, from, to strin
 		return ErrServiceImportAuthorization
 	}
 
+	if a.importFormsCycle(destination, from, to) {
+		return ErrServiceImportFormsCycle
+	}
+
 	_, err := a.addServiceImport(destination, from, to, imClaim)
 	return err
 }
 
+// Detects if we have a cycle.
+func (a *Account) importFormsCycle(destination *Account, from, to string) bool {
+	// Check that what we are importing is not something we also export.
+	if a.serviceExportOverlaps(to) {
+		// So at this point if destination account is also importing from us, that forms a cycle.
+		if destination.serviceImportOverlaps(from) {
+			return true
+		}
+	}
+	return false
+}
+
 // SetServiceImportSharing will allow sharing of information about requests with the export account.
 // Used for service latency tracking at the moment.
 func (a *Account) SetServiceImportSharing(destination *Account, to string, allow bool) error {
@@ -1586,6 +1602,30 @@ func (a *Account) checkForReverseEntry(reply string, si *serviceImport, checkInt
 	}
 }
 
+// Internal check to see if the to subject overlaps with another export.
+func (a *Account) serviceExportOverlaps(to string) bool {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	for subj := range a.exports.services {
+		if to == subj || SubjectsCollide(to, subj) {
+			return true
+		}
+	}
+	return false
+}
+
+// Internal check to see if the from subject overlaps with another import.
+func (a *Account) serviceImportOverlaps(from string) bool {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	for subj := range a.imports.services {
+		if from == subj || SubjectsCollide(from, subj) {
+			return true
+		}
+	}
+	return false
+}
+
 // Internal check to see if a service import exists.
 func (a *Account) serviceImportExists(dest *Account, from string) bool {
 	a.mu.RLock()
diff --git a/server/client.go b/server/client.go
index 60c5e26c..11fe6701 100644
--- a/server/client.go
+++ b/server/client.go
@@ -177,6 +177,7 @@ const (
 	MsgHeaderViolation
 	NoRespondersRequiresHeaders
 	ClusterNameConflict
+	DuplicateRemoteLeafnodeConnection
 )
 
 // Some flags passed to processMsgResults
@@ -3719,6 +3720,7 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 			}
 			continue
 		}
+
 		// Assume delivery subject is the normal subject to this point.
 		dsubj = subj
 
diff --git a/server/errors.go b/server/errors.go
index ef085acf..28f12fde 100644
--- a/server/errors.go
+++ b/server/errors.go
@@ -125,6 +125,9 @@ var (
 	// ErrServiceImportAuthorization is returned when a service import is not authorized.
 	ErrServiceImportAuthorization = errors.New("service import not authorized")
 
+	// ErrServiceImportFormsCycle is returned when a service import forms a cycle.
+	ErrServiceImportFormsCycle = errors.New("service import forms cycle")
+
 	// ErrClientOrRouteConnectedToGatewayPort represents an error condition when
 	// a client or route attempted to connect to the Gateway port.
 	ErrClientOrRouteConnectedToGatewayPort = errors.New("attempted to connect to gateway port")
diff --git a/server/leafnode.go b/server/leafnode.go
index f6b21fd4..61264f77 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -125,6 +125,56 @@ func validateLeafNode(o *Options) error {
 	if err := validateLeafNodeAuthOptions(o); err != nil {
 		return err
 	}
+	// In local config mode, check that leafnode configuration refers to accounts that exist.
+	if len(o.TrustedOperators) == 0 {
+		accNames := map[string]struct{}{}
+		for _, a := range o.Accounts {
+			accNames[a.Name] = struct{}{}
+		}
+		// global account is always created
+		accNames[DEFAULT_GLOBAL_ACCOUNT] = struct{}{}
+		// in the context of leaf nodes, empty account means global account
+		accNames[_EMPTY_] = struct{}{}
+		// system account either exists or, if not disabled, will be created
+		if o.SystemAccount == _EMPTY_ && !o.NoSystemAccount {
+			accNames[DEFAULT_SYSTEM_ACCOUNT] = struct{}{}
+		}
+		checkAccountExists := func(accName string, cfgType string) error {
+			if _, ok := accNames[accName]; !ok {
+				return fmt.Errorf("cannot find local account %q specified in leafnode %s", accName, cfgType)
+			}
+			return nil
+		}
+		if err := checkAccountExists(o.LeafNode.Account, "authorization"); err != nil {
+			return err
+		}
+		for _, lu := range o.LeafNode.Users {
+			if lu.Account == nil { // means global account
+				continue
+			}
+			if err := checkAccountExists(lu.Account.Name, "authorization"); err != nil {
+				return err
+			}
+		}
+		for _, r := range o.LeafNode.Remotes {
+			if err := checkAccountExists(r.LocalAccount, "remote"); err != nil {
+				return err
+			}
+		}
+	} else {
+		if len(o.LeafNode.Users) != 0 {
+			return fmt.Errorf("operator mode does not allow specifying user in leafnode config")
+		}
+		for _, r := range o.LeafNode.Remotes {
+			if !nkeys.IsValidPublicAccountKey(r.LocalAccount) {
+				return fmt.Errorf("operator mode requires account nkeys in remotes")
+			}
+		}
+		if o.LeafNode.Port != 0 && o.LeafNode.Account != "" && !nkeys.IsValidPublicAccountKey(o.LeafNode.Account) {
+			return fmt.Errorf("operator mode and non account nkeys are incompatible")
+		}
+	}
+
 	if o.LeafNode.Port == 0 {
 		return nil
 	}
@@ -836,7 +886,7 @@ func (s *Server) createLeafNode(conn net.Conn, remote *leafNodeCfg) *client {
 	if solicited {
 		// Make sure we register with the account here.
 		c.registerWithAccount(acc)
-		s.addLeafNodeConnection(c)
+		s.addLeafNodeConnection(c, _EMPTY_, false)
 		s.initLeafNodeSmapAndSendSubs(c)
 		if sendSysConnectEvent {
 			s.sendLeafNodeConnect(acc)
@@ -1006,14 +1056,53 @@ func (s *Server) setLeafNodeInfoHostPortAndIP() error {
 	return nil
 }
 
-func (s *Server) addLeafNodeConnection(c *client) {
+// Add the connection to the map of leaf nodes.
+// If `checkForDup` is true (invoked when a leafnode is accepted), then we check
+// if a connection already exists for the same server name (ID) and account.
+// That can happen when the remote is attempting to reconnect while the accepting
+// side did not detect the connection as broken yet.
+// But it can also happen when there is a misconfiguration and the remote is
+// creating two (or more) connections that bind to the same account on the accept
+// side.
+// When a duplicate is found, the new connection is accepted and the old is closed
+// (this solves the stale connection situation). An error is returned to help the
+// remote detect the misconfiguration when the duplicate is the result of that
+// misconfiguration.
+func (s *Server) addLeafNodeConnection(c *client, srvName string, checkForDup bool) {
+	var accName string
 	c.mu.Lock()
 	cid := c.cid
+	if c.acc != nil {
+		accName = c.acc.Name
+	}
 	c.mu.Unlock()
+
+	var old *client
 	s.mu.Lock()
+	if checkForDup {
+		for _, ol := range s.leafs {
+			ol.mu.Lock()
+			// We check for empty because in some test we may send empty CONNECT{}
+			if srvName != _EMPTY_ && ol.opts.Name == srvName && ol.acc.Name == accName {
+				old = ol
+			}
+			ol.mu.Unlock()
+			if old != nil {
+				break
+			}
+		}
+	}
+	// Store new connection in the map
 	s.leafs[cid] = c
 	s.mu.Unlock()
 	s.removeFromTempClients(cid)
+
+	// If applicable, evict the old one.
+	if old != nil {
+		old.sendErrAndErr(DuplicateRemoteLeafnodeConnection.String())
+		old.closeConnection(DuplicateRemoteLeafnodeConnection)
+		c.Warnf("Replacing connection from same server")
+	}
 }
 
 func (s *Server) removeLeafNodeConnection(c *client) {
@@ -1073,9 +1162,7 @@ func (c *client) processLeafNodeConnect(s *Server, arg []byte, lang string) erro
 		return ErrWrongGateway
 	}
 
-	// Check for stale connection from same server/account
-	c.replaceOldLeafNodeConnIfNeeded(s, proto)
-
+	c.mu.Lock()
 	// Leaf Nodes do not do echo or verbose or pedantic.
 	c.opts.Verbose = false
 	c.opts.Echo = false
@@ -1090,6 +1177,10 @@ func (c *client) processLeafNodeConnect(s *Server, arg []byte, lang string) erro
 	if proto.Cluster != "" {
 		c.leaf.remoteCluster = proto.Cluster
 	}
+	c.mu.Unlock()
+
+	// Add in the leafnode here since we passed through auth at this point.
+	s.addLeafNodeConnection(c, proto.Name, true)
 
 	// If we have permissions bound to this leafnode we need to send then back to the
 	// origin server for local enforcement.
@@ -1099,9 +1190,6 @@ func (c *client) processLeafNodeConnect(s *Server, arg []byte, lang string) erro
 	// This will send all registered subs too.
 	s.initLeafNodeSmapAndSendSubs(c)
 
-	// Add in the leafnode here since we passed through auth at this point.
-	s.addLeafNodeConnection(c)
-
 	// Announce the account connect event for a leaf node.
 	// This will no-op as needed.
 	s.sendLeafNodeConnect(c.acc)
@@ -1109,42 +1197,6 @@ func (c *client) processLeafNodeConnect(s *Server, arg []byte, lang string) erro
 	return nil
 }
 
-// Invoked from a server accepting a leafnode connection. It looks for a possible
-// existing leafnode connection from the same server with the same account, and
-// if it finds one, closes it so that the new one is accepted and not reported as
-// forming a cycle.
-//
-// This must be invoked for LEAF client types, and on the server accepting the connection.
-//
-// No server nor client lock held on entry.
-func (c *client) replaceOldLeafNodeConnIfNeeded(s *Server, connInfo *leafConnectInfo) {
-	var accName string
-	c.mu.Lock()
-	if c.acc != nil {
-		accName = c.acc.Name
-	}
-	c.mu.Unlock()
-
-	var old *client
-	s.mu.Lock()
-	for _, ol := range s.leafs {
-		ol.mu.Lock()
-		// We check for empty because in some test we may send empty CONNECT{}
-		if ol.opts.Name == connInfo.Name && connInfo.Name != _EMPTY_ && ol.acc.Name == accName {
-			old = ol
-		}
-		ol.mu.Unlock()
-		if old != nil {
-			break
-		}
-	}
-	s.mu.Unlock()
-	if old != nil {
-		old.Warnf("Replacing connection from same server")
-		old.closeConnection(ReadError)
-	}
-}
-
 // Returns the remote cluster name. This is set only once so does not require a lock.
 func (c *client) remoteCluster() string {
 	if c.leaf == nil {
@@ -1322,7 +1374,10 @@ func (s *Server) updateLeafNodes(acc *Account, sub *subscription, delta int32) {
 
 	for _, ln := range leafs {
 		// Check to make sure this sub does not have an origin cluster than matches the leafnode.
-		if sub.origin != nil && string(sub.origin) == ln.remoteCluster() {
+		ln.mu.Lock()
+		skip := sub.origin != nil && string(sub.origin) == ln.remoteCluster()
+		ln.mu.Unlock()
+		if skip {
 			continue
 		}
 		ln.updateSmap(sub, delta)
diff --git a/server/monitor.go b/server/monitor.go
index a1e3858b..4bae0fda 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -1913,6 +1913,8 @@ func (reason ClosedState) String() string {
 		return "No Responders Requires Headers"
 	case ClusterNameConflict:
 		return "Cluster Name Conflict"
+	case DuplicateRemoteLeafnodeConnection:
+		return "Duplicate Remote LeafNode Connection"
 	}
 
 	return "Unknown State"
diff --git a/server/parser.go b/server/parser.go
index d8b7f87e..d7a0d73f 100644
--- a/server/parser.go
+++ b/server/parser.go
@@ -132,6 +132,7 @@ const (
 func (c *client) parse(buf []byte) error {
 	var i int
 	var b byte
+	var lmsg bool
 
 	// Snapshots
 	c.mu.Lock()
@@ -463,6 +464,7 @@ func (c *client) parse(buf []byte) error {
 			// Drop all pub args
 			c.pa.arg, c.pa.pacache, c.pa.origin, c.pa.account, c.pa.subject = nil, nil, nil, nil, nil
 			c.pa.reply, c.pa.hdr, c.pa.size, c.pa.szb, c.pa.hdb, c.pa.queues = nil, -1, 0, nil, nil, nil
+			lmsg = false
 		case OP_A:
 			switch b {
 			case '+':
@@ -932,6 +934,7 @@ func (c *client) parse(buf []byte) error {
 						if trace {
 							c.traceInOp("LMSG", arg)
 						}
+						lmsg = true
 						err = c.processRoutedOriginClusterMsgArgs(arg)
 					}
 				} else if c.kind == LEAF {
@@ -1114,7 +1117,7 @@ func (c *client) parse(buf []byte) error {
 
 		if c.argBuf == nil {
 			// Works also for MSG_ARG, when message comes from ROUTE.
-			if err := c.clonePubArg(); err != nil {
+			if err := c.clonePubArg(lmsg); err != nil {
 				goto parseErr
 			}
 		}
@@ -1165,13 +1168,16 @@ func protoSnippet(start int, buf []byte) string {
 
 // clonePubArg is used when the split buffer scenario has the pubArg in the existing read buffer, but
 // we need to hold onto it into the next read.
-func (c *client) clonePubArg() error {
+func (c *client) clonePubArg(lmsg bool) error {
 	// Just copy and re-process original arg buffer.
 	c.argBuf = c.scratch[:0]
 	c.argBuf = append(c.argBuf, c.pa.arg...)
 
 	switch c.kind {
 	case ROUTER, GATEWAY:
+		if lmsg {
+			return c.processRoutedOriginClusterMsgArgs(c.argBuf)
+		}
 		if c.pa.hdr < 0 {
 			return c.processRoutedMsgArgs(c.argBuf)
 		} else {
diff --git a/server/pse/freebsd.txt b/server/pse/freebsd.txt
new file mode 100644
index 00000000..877068cc
--- /dev/null
+++ b/server/pse/freebsd.txt
@@ -0,0 +1,30 @@
+/*
+ * Compile and run this as a C program to get the kinfo_proc offsets
+ * for your architecture.
+ * While FreeBSD works hard at binary-compatibility within an ABI, various
+ * we can't say for sure that these are right for _all_ use on a hardware
+ * platform.  The LP64 ifdef affects the offsets considerably.
+ *
+ * We use these offsets in hardware-specific files for FreeBSD, to avoid a cgo
+ * compilation-time dependency, allowing us to cross-compile for FreeBSD from
+ * other hardware platforms, letting us distribute binaries for FreeBSD.
+ */
+
+#include <stddef.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <sys/types.h>
+#include <sys/user.h>
+
+#define SHOW_OFFSET(FIELD) printf(" KIP_OFF_%s = %zu\n", #FIELD, offsetof(struct kinfo_proc, ki_ ## FIELD))
+
+int main(int argc, char *argv[]) {
+	/* Uncomment these if you want some extra debugging aids:
+	SHOW_OFFSET(pid);
+	SHOW_OFFSET(ppid);
+	SHOW_OFFSET(uid);
+	*/
+	SHOW_OFFSET(size);
+	SHOW_OFFSET(rssize);
+	SHOW_OFFSET(pctcpu);
+}
diff --git a/server/pse/pse_freebsd.go b/server/pse/pse_freebsd.go
index 40c52847..849440fc 100644
--- a/server/pse/pse_freebsd.go
+++ b/server/pse/pse_freebsd.go
@@ -11,6 +11,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+// +build !amd64
+
 package pse
 
 /*
@@ -54,7 +56,7 @@ int getusage(double *pcpu, unsigned int *rss, unsigned int *vss)
 
     *rss = pagetok(kp.ki_rssize);
     *vss = kp.ki_size;
-    *pcpu = kp.ki_pctcpu;
+    *pcpu = (double)kp.ki_pctcpu / FSCALE;
 
     return 0;
 }
diff --git a/server/pse/pse_freebsd_amd64.go b/server/pse/pse_freebsd_amd64.go
new file mode 100644
index 00000000..2d21e5c2
--- /dev/null
+++ b/server/pse/pse_freebsd_amd64.go
@@ -0,0 +1,91 @@
+// Copyright 2015-2020 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// This is the amd64-specific FreeBSD implementation, with hard-coded offset
+// constants derived by running freebsd.txt; having this implementation allows
+// us to compile without CGO, which lets us cross-compile for FreeBSD from our
+// CI system and so supply binaries for FreeBSD amd64.
+//
+// To generate for other architectures:
+//   1. Update pse_freebsd.go, change the build exclusion to exclude your arch
+//   2. Copy this file to be built for your arch
+//   3. Update `nativeEndian` below
+//   4. Link `freebsd.txt` to have a .c filename and compile and run, then
+//      paste the outputs into the const section below.
+
+package pse
+
+import (
+	"encoding/binary"
+	"syscall"
+
+	"golang.org/x/sys/unix"
+)
+
+// On FreeBSD, to get proc information we read it out of the kernel using a
+// binary sysctl.  The endianness of integers is thus explicitly "host", rather
+// than little or big endian.
+var nativeEndian = binary.LittleEndian
+
+const (
+	KIP_OFF_size   = 256
+	KIP_OFF_rssize = 264
+	KIP_OFF_pctcpu = 308
+)
+
+var pageshift int
+
+func init() {
+	// To get the physical page size, the C library checks two places:
+	//   process ELF auxiliary info, AT_PAGESZ
+	//   as a fallback, the hw.pagesize sysctl
+	// In looking closely, I found that the Go runtime support is handling
+	// this for us, and exposing that as syscall.Getpagesize, having checked
+	// both in the same ways, at process start, so a call to that should return
+	// a memory value without even a syscall bounce.
+	pagesize := syscall.Getpagesize()
+	pageshift = 0
+	for pagesize > 1 {
+		pageshift += 1
+		pagesize >>= 1
+	}
+}
+
+func ProcUsage(pcpu *float64, rss, vss *int64) error {
+	rawdata, err := unix.SysctlRaw("kern.proc.pid", unix.Getpid())
+	if err != nil {
+		return err
+	}
+
+	r_vss_bytes := nativeEndian.Uint32(rawdata[KIP_OFF_size:])
+	r_rss_pages := nativeEndian.Uint32(rawdata[KIP_OFF_rssize:])
+	rss_bytes := r_rss_pages << pageshift
+
+	// In C: fixpt_t ki_pctcpu
+	// Doc: %cpu for process during ki_swtime
+	// fixpt_t is __uint32_t
+	// usr.bin/top uses pctdouble to convert to a double (float64)
+	// define pctdouble(p) ((double)(p) / FIXED_PCTCPU)
+	// FIXED_PCTCPU is _usually_ FSCALE (some architectures are special)
+	// <sys/param.h> has:
+	//   #define FSHIFT  11              /* bits to right of fixed binary point */
+	//   #define FSCALE  (1<<FSHIFT)
+	r_pcpu := nativeEndian.Uint32(rawdata[KIP_OFF_pctcpu:])
+	f_pcpu := float64(r_pcpu) / float64(2048)
+
+	*rss = int64(rss_bytes)
+	*vss = int64(r_vss_bytes)
+	*pcpu = f_pcpu
+
+	return nil
+}
diff --git a/server/server.go b/server/server.go
index 1811c8d3..455457a7 100644
--- a/server/server.go
+++ b/server/server.go
@@ -414,45 +414,6 @@ func NewServer(opts *Options) (*Server, error) {
 		return nil, err
 	}
 
-	// In local config mode, check that leafnode configuration
-	// refers to account that exist.
-	if len(opts.TrustedOperators) == 0 {
-		checkAccountExists := func(accName string) error {
-			if accName == _EMPTY_ {
-				return nil
-			}
-			if _, ok := s.accounts.Load(accName); !ok {
-				if accName == DEFAULT_SYSTEM_ACCOUNT && opts.SystemAccount == _EMPTY_ && !opts.NoSystemAccount {
-					return nil
-				}
-				return fmt.Errorf("cannot find account %q specified in leafnode authorization", accName)
-			}
-			return nil
-		}
-		if err := checkAccountExists(opts.LeafNode.Account); err != nil {
-			return nil, err
-		}
-		for _, lu := range opts.LeafNode.Users {
-			if lu.Account == nil {
-				continue
-			}
-			if err := checkAccountExists(lu.Account.Name); err != nil {
-				return nil, err
-			}
-		}
-		for _, r := range opts.LeafNode.Remotes {
-			if r.LocalAccount == _EMPTY_ {
-				continue
-			}
-			if _, ok := s.accounts.Load(r.LocalAccount); !ok {
-				if r.LocalAccount == DEFAULT_SYSTEM_ACCOUNT && opts.SystemAccount == _EMPTY_ && !opts.NoSystemAccount {
-					continue
-				}
-				return nil, fmt.Errorf("no local account %q for remote leafnode", r.LocalAccount)
-			}
-		}
-	}
-
 	// Used to setup Authorization.
 	s.configureAuthorization()
 
diff --git a/server/stream.go b/server/stream.go
index 2698080a..60a72178 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -52,11 +52,19 @@ type StreamConfig struct {
 	Duplicates   time.Duration   `json:"duplicate_window,omitempty"`
 }
 
+const JSApiPubAckResponseType = "io.nats.jetstream.api.v1.pub_ack_response"
+
+// JSPubAckResponse is a formal response to a publish operation.
+type JSPubAckResponse struct {
+	ApiResponse
+	*PubAck
+}
+
 // PubAck is the detail you get back from a publish to a stream that was successful.
 // e.g. +OK {"stream": "Orders", "seq": 22}
 type PubAck struct {
 	Stream    string `json:"stream"`
-	Seq       uint64 `json:"seq"`
+	Sequence  uint64 `json:"seq"`
 	Duplicate bool   `json:"duplicate,omitempty"`
 }
 
@@ -77,6 +85,8 @@ type Stream struct {
 	pubAck    []byte
 	sendq     chan *jsPubMsg
 	store     StreamStore
+	lseq      uint64
+	lmsgId    string
 	consumers map[string]*Consumer
 	numFilter int
 	config    StreamConfig
@@ -87,9 +97,13 @@ type Stream struct {
 	ddtmr     *time.Timer
 }
 
-// JSPubId is used for identifying published messages and performing de-duplication.
-const JSPubId = "Msg-Id"
-const StreamDefaultDuplicatesWindow = 2 * time.Minute
+// Headers for published messages.
+const (
+	JSMsgId             = "Nats-Msg-Id"
+	JSExpectedStream    = "Nats-Expected-Stream"
+	JSExpectedLastSeq   = "Nats-Expected-Last-Sequence"
+	JSExpectedLastMsgId = "Nats-Expected-Last-Msg-Id"
+)
 
 // Dedupe entry
 type ddentry struct {
@@ -181,13 +195,14 @@ func (a *Account) AddStreamWithStore(config *StreamConfig, fsConfig *FileStoreCo
 	// Setup our internal send go routine.
 	mset.setupSendCapabilities()
 
-	// Create our pubAck here. This will be reused and for +OK will contain JSON
-	// for stream name and sequence.
-	longestSeq := strconv.FormatUint(math.MaxUint64, 10)
-	lpubAck := len(OK) + len(cfg.Name) + len("{\"stream\": ,\"seq\": }") + len(longestSeq)
-	mset.pubAck = make([]byte, 0, lpubAck)
-	mset.pubAck = append(mset.pubAck, OK...)
-	mset.pubAck = append(mset.pubAck, fmt.Sprintf(" {\"stream\": %q, \"seq\": ", cfg.Name)...)
+	// Create our pubAck template here. Better than json marshal each time on success.
+	b, _ := json.Marshal(&JSPubAckResponse{
+		ApiResponse: ApiResponse{Type: JSApiPubAckResponseType},
+		PubAck:      &PubAck{Stream: cfg.Name, Sequence: math.MaxUint64},
+	})
+	end := bytes.Index(b, []byte(strconv.FormatUint(math.MaxUint64, 10)))
+	// We need to force cap here to make sure this is a copy when sending a response.
+	mset.pubAck = b[:end:end]
 
 	// Rebuild dedupe as needed.
 	mset.rebuildDedupe()
@@ -270,14 +285,12 @@ func (mset *Stream) autoTuneFileStorageBlockSize(fsCfg *FileStoreConfig) {
 }
 
 // rebuildDedupe will rebuild any dedupe structures needed after recovery of a stream.
-// Lock not needed, only called during initialization.
 // TODO(dlc) - Might be good to know if this should be checked at all for streams with no
 // headers and msgId in them. Would need signaling from the storage layer.
 func (mset *Stream) rebuildDedupe() {
 	state := mset.store.State()
-	if state.Msgs == 0 {
-		return
-	}
+	mset.lseq = state.LastSeq
+
 	// We have some messages. Lookup starting sequence by duplicate time window.
 	sseq := mset.store.GetSeqFromTime(time.Now().Add(-mset.config.Duplicates))
 	if sseq == 0 {
@@ -286,11 +299,15 @@ func (mset *Stream) rebuildDedupe() {
 
 	for seq := sseq; seq <= state.LastSeq; seq++ {
 		_, hdr, _, ts, err := mset.store.LoadMsg(seq)
+		var msgId string
 		if err == nil && len(hdr) > 0 {
-			if msgId := getMsgId(hdr); msgId != "" {
+			if msgId = getMsgId(hdr); msgId != _EMPTY_ {
 				mset.storeMsgId(&ddentry{msgId, seq, ts})
 			}
 		}
+		if seq == state.LastSeq {
+			mset.lmsgId = msgId
+		}
 	}
 }
 
@@ -399,6 +416,9 @@ func (jsa *jsAccount) subjectsOverlap(subjects []string) bool {
 	return false
 }
 
+// Default duplicates window.
+const StreamDefaultDuplicatesWindow = 2 * time.Minute
+
 func checkStreamCfg(config *StreamConfig) (StreamConfig, error) {
 	if config == nil {
 		return StreamConfig{}, fmt.Errorf("stream configuration invalid")
@@ -861,7 +881,26 @@ func getHdrVal(key string, hdr []byte) []byte {
 
 // Fast lookup of msgId.
 func getMsgId(hdr []byte) string {
-	return string(getHdrVal(JSPubId, hdr))
+	return string(getHdrVal(JSMsgId, hdr))
+}
+
+// Fast lookup of expected last msgId.
+func getExpectedLastMsgId(hdr []byte) string {
+	return string(getHdrVal(JSExpectedLastMsgId, hdr))
+}
+
+// Fast lookup of expected stream.
+func getExpectedStream(hdr []byte) string {
+	return string(getHdrVal(JSExpectedStream, hdr))
+}
+
+// Fast lookup of expected stream.
+func getExpectedLastSeq(hdr []byte) uint64 {
+	bseq := getHdrVal(JSExpectedLastSeq, hdr)
+	if len(bseq) == 0 {
+		return 0
+	}
+	return uint64(parseInt64(bseq))
 }
 
 // processInboundJetStreamMsg handles processing messages bound for a stream.
@@ -883,17 +922,53 @@ func (mset *Stream) processInboundJetStreamMsg(_ *subscription, pc *client, subj
 	numConsumers := len(mset.consumers)
 	interestRetention := mset.config.Retention == InterestPolicy
 
-	// Process msgId if we have headers.
+	var resp = &JSPubAckResponse{ApiResponse: ApiResponse{Type: JSApiPubAckResponseType}}
+
+	// Process msg headers if present.
 	var msgId string
 	if pc != nil && pc.pa.hdr > 0 {
-		msgId = getMsgId(msg[:pc.pa.hdr])
+		hdr := msg[:pc.pa.hdr]
+		msgId = getMsgId(hdr)
+		sendq := mset.sendq
 		if dde := mset.checkMsgId(msgId); dde != nil {
+			mset.mu.Unlock()
 			if doAck && len(reply) > 0 {
 				response := append(pubAck, strconv.FormatUint(dde.seq, 10)...)
-				response = append(response, ", \"duplicate\": true}"...)
-				mset.sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, response, nil, 0}
+				response = append(response, ",\"duplicate\": true}"...)
+				sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, response, nil, 0}
 			}
+			return
+		}
+		// Expected stream.
+		if sname := getExpectedStream(hdr); sname != _EMPTY_ && sname != name {
 			mset.mu.Unlock()
+			if doAck && len(reply) > 0 {
+				resp.Error = &ApiError{Code: 400, Description: "expected stream does not match"}
+				b, _ := json.Marshal(resp)
+				sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, b, nil, 0}
+			}
+			return
+		}
+		// Expected last sequence.
+		if seq := getExpectedLastSeq(hdr); seq > 0 && seq != mset.lseq {
+			lseq := mset.lseq
+			mset.mu.Unlock()
+			if doAck && len(reply) > 0 {
+				resp.Error = &ApiError{Code: 400, Description: fmt.Sprintf("wrong last sequence: %d", lseq)}
+				b, _ := json.Marshal(resp)
+				sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, b, nil, 0}
+			}
+			return
+		}
+		// Expected last msgId.
+		if lmsgId := getExpectedLastMsgId(hdr); lmsgId != _EMPTY_ && lmsgId != mset.lmsgId {
+			last := mset.lmsgId
+			mset.mu.Unlock()
+			if doAck && len(reply) > 0 {
+				resp.Error = &ApiError{Code: 400, Description: fmt.Sprintf("wrong last msg ID: %s", last)}
+				b, _ := json.Marshal(resp)
+				sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, b, nil, 0}
+			}
 			return
 		}
 	}
@@ -917,9 +992,10 @@ func (mset *Stream) processInboundJetStreamMsg(_ *subscription, pc *client, subj
 	// Check to see if we are over the max msg size.
 	if maxMsgSize >= 0 && len(msg) > maxMsgSize {
 		mset.mu.Unlock()
-		response = []byte("-ERR 'message size exceeds maximum allowed'")
 		if doAck && len(reply) > 0 {
-			mset.sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, response, nil, 0}
+			resp.Error = &ApiError{Code: 400, Description: "message size exceeds maximum allowed"}
+			b, _ := json.Marshal(resp)
+			mset.sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, b, nil, 0}
 		}
 		return
 	}
@@ -941,13 +1017,15 @@ func (mset *Stream) processInboundJetStreamMsg(_ *subscription, pc *client, subj
 			}
 		}
 	}
-	mset.mu.Unlock()
 
-	// Skip here.
+	// Skip msg here.
 	if noInterest {
-		seq = store.SkipMsg()
+		mset.lseq = store.SkipMsg()
+		mset.lmsgId = msgId
+		mset.mu.Unlock()
+
 		if doAck && len(reply) > 0 {
-			response = append(pubAck, strconv.FormatUint(seq, 10)...)
+			response = append(pubAck, strconv.FormatUint(mset.lseq, 10)...)
 			response = append(response, '}')
 			mset.sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, response, nil, 0}
 		}
@@ -959,31 +1037,45 @@ func (mset *Stream) processInboundJetStreamMsg(_ *subscription, pc *client, subj
 	}
 
 	// If here we will attempt to store the message.
-	// Headers.
+	// Check for headers.
 	if pc != nil && pc.pa.hdr > 0 {
 		hdr = msg[:pc.pa.hdr]
 		msg = msg[pc.pa.hdr:]
 	}
 	seq, ts, err = store.StoreMsg(subject, hdr, msg)
+	if err == nil && seq > 0 {
+		mset.lseq = seq
+		mset.lmsgId = msgId
+	}
+
+	// We hold the lock to this point to make sure nothing gets between us since we check for pre-conditions.
+	mset.mu.Unlock()
+
 	if err != nil {
 		if err != ErrStoreClosed {
 			c.Errorf("JetStream failed to store a msg on account: %q stream: %q -  %v", accName, name, err)
 		}
-		response = []byte(fmt.Sprintf("-ERR '%v'", err))
+		if doAck && len(reply) > 0 {
+			resp.Error = &ApiError{Code: 400, Description: err.Error()}
+			response, _ = json.Marshal(resp)
+		}
 	} else if jsa.limitsExceeded(stype) {
 		c.Warnf("JetStream resource limits exceeded for account: %q", accName)
-		response = []byte("-ERR 'resource limits exceeded for account'")
+		if doAck && len(reply) > 0 {
+			resp.Error = &ApiError{Code: 400, Description: "resource limits exceeded for account"}
+			response, _ = json.Marshal(resp)
+		}
 		store.RemoveMsg(seq)
 		seq = 0
 	} else {
-		if doAck && len(reply) > 0 {
-			response = append(pubAck, strconv.FormatUint(seq, 10)...)
-			response = append(response, '}')
-		}
 		// If we have a msgId make sure to save.
 		if msgId != "" {
 			mset.storeMsgId(&ddentry{msgId, seq, ts})
 		}
+		if doAck && len(reply) > 0 {
+			response = append(pubAck, strconv.FormatUint(seq, 10)...)
+			response = append(response, '}')
+		}
 	}
 
 	// Send response here.


Test patch:
diff --git a/server/leafnode_test.go b/server/leafnode_test.go
index e87b24d0..cfbf1c69 100644
--- a/server/leafnode_test.go
+++ b/server/leafnode_test.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"io/ioutil"
 	"net"
 	"net/url"
 	"os"
@@ -1803,6 +1804,68 @@ func TestLeafNodeLoopDetectedDueToReconnect(t *testing.T) {
 	checkLeafNodeConnected(t, sl)
 }
 
+func TestLeafNodeTwoRemotesBindToSameAccount(t *testing.T) {
+	opts := DefaultOptions()
+	opts.LeafNode.Host = "127.0.0.1"
+	opts.LeafNode.Port = -1
+	s := RunServer(opts)
+	defer s.Shutdown()
+
+	conf := `
+	listen: 127.0.0.1:-1
+	cluster { name: ln22, listen: 127.0.0.1:-1 }
+	accounts {
+		a { users [ {user: a, password: a} ]}
+		b { users [ {user: b, password: b} ]}
+	}
+	leafnodes {
+		remotes = [
+			{
+				url: nats-leaf://127.0.0.1:%d
+				account: a
+			}
+			{
+				url: nats-leaf://127.0.0.1:%d
+				account: b
+			}
+		]
+	}
+	`
+	lconf := createConfFile(t, []byte(fmt.Sprintf(conf, opts.LeafNode.Port, opts.LeafNode.Port)))
+	defer os.Remove(lconf)
+
+	lopts, err := ProcessConfigFile(lconf)
+	if err != nil {
+		t.Fatalf("Error loading config file: %v", err)
+	}
+	lopts.NoLog = false
+	ln, err := NewServer(lopts)
+	if err != nil {
+		t.Fatalf("Error creating server: %v", err)
+	}
+	defer ln.Shutdown()
+	l := &captureErrorLogger{errCh: make(chan string, 10)}
+	ln.SetLogger(l, false, false)
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		ln.Start()
+	}()
+
+	select {
+	case err := <-l.errCh:
+		if !strings.Contains(err, DuplicateRemoteLeafnodeConnection.String()) {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("Did not get any error")
+	}
+	ln.Shutdown()
+	wg.Wait()
+}
+
 func TestLeafNodeNoDuplicateWithinCluster(t *testing.T) {
 	// This set the cluster name to "abc"
 	oSrv1 := DefaultOptions()
@@ -1883,3 +1946,125 @@ func TestLeafNodeNoDuplicateWithinCluster(t *testing.T) {
 		}
 	}
 }
+
+func TestLeafNodeNoRouteParserError(t *testing.T) {
+	// This set the cluster name to "abc"
+	oSrv1 := DefaultOptions()
+	oSrv1.LeafNode.Host = "127.0.0.1"
+	oSrv1.LeafNode.Port = -1
+	srv1 := RunServer(oSrv1)
+	defer srv1.Shutdown()
+
+	oSrv2 := DefaultOptions()
+	oSrv2.LeafNode.Host = "127.0.0.1"
+	oSrv2.LeafNode.Port = -1
+	oSrv2.Routes = RoutesFromStr(fmt.Sprintf("nats://127.0.0.1:%d", oSrv1.Cluster.Port))
+	srv2 := RunServer(oSrv2)
+	defer srv2.Shutdown()
+
+	checkClusterFormed(t, srv1, srv2)
+
+	u1, err := url.Parse(fmt.Sprintf("nats://127.0.0.1:%d", oSrv1.LeafNode.Port))
+	if err != nil {
+		t.Fatalf("Error parsing url: %v", err)
+	}
+	u2, err := url.Parse(fmt.Sprintf("nats://127.0.0.1:%d", oSrv2.LeafNode.Port))
+	if err != nil {
+		t.Fatalf("Error parsing url: %v", err)
+	}
+	remoteLeafs := []*RemoteLeafOpts{&RemoteLeafOpts{URLs: []*url.URL{u1, u2}}}
+
+	oLeaf1 := DefaultOptions()
+	oLeaf1.LeafNode.Remotes = remoteLeafs
+	leaf1 := RunServer(oLeaf1)
+	defer leaf1.Shutdown()
+
+	oLeaf2 := DefaultOptions()
+	oLeaf2.LeafNode.Remotes = remoteLeafs
+	oLeaf2.Routes = RoutesFromStr(fmt.Sprintf("nats://127.0.0.1:%d", oLeaf1.Cluster.Port))
+	leaf2 := RunServer(oLeaf2)
+	defer leaf2.Shutdown()
+
+	checkClusterFormed(t, leaf1, leaf2)
+
+	checkLeafNodeConnected(t, leaf1)
+	checkLeafNodeConnected(t, leaf2)
+
+	ncSrv2 := natsConnect(t, srv2.ClientURL())
+	defer ncSrv2.Close()
+	natsQueueSub(t, ncSrv2, "foo", "queue", func(m *nats.Msg) {
+		m.Respond([]byte("from srv2"))
+	})
+
+	// Check that "foo" interest is available everywhere.
+	checkSubInterest(t, srv1, globalAccountName, "foo", time.Second)
+	checkSubInterest(t, srv2, globalAccountName, "foo", time.Second)
+	checkSubInterest(t, leaf1, globalAccountName, "foo", time.Second)
+	checkSubInterest(t, leaf2, globalAccountName, "foo", time.Second)
+
+	// Not required, but have a request payload that is more than 100 bytes
+	reqPayload := make([]byte, 150)
+	for i := 0; i < len(reqPayload); i++ {
+		reqPayload[i] = byte((i % 26)) + 'A'
+	}
+
+	// Send repeated requests (from scratch) from leaf-2:
+	sendReq := func() {
+		t.Helper()
+
+		ncLeaf2 := natsConnect(t, leaf2.ClientURL())
+		defer ncLeaf2.Close()
+
+		if _, err := ncLeaf2.Request("foo", reqPayload, time.Second); err != nil {
+			t.Fatalf("Did not receive reply: %v", err)
+		}
+	}
+	for i := 0; i < 100; i++ {
+		sendReq()
+	}
+}
+
+func TestLeafNodeOperatorBadCfg(t *testing.T) {
+	tmpDir, err := ioutil.TempDir("", "_nats-server")
+	if err != nil {
+		t.Fatal("Could not create tmp dir")
+	}
+	defer os.RemoveAll(tmpDir)
+	for errorText, cfg := range map[string]string{
+		"operator mode does not allow specifying user in leafnode config": `
+			port: -1
+			authorization {
+				users = [{user: "u", password: "p"}]}
+			}`,
+		`operator mode and non account nkeys are incompatible`: `
+			port: -1
+			authorization {
+				account: notankey
+			}`,
+		`operator mode requires account nkeys in remotes`: `remotes: [{url: u}]`,
+	} {
+		t.Run(errorText, func(t *testing.T) {
+			conf := createConfFile(t, []byte(fmt.Sprintf(`
+		port: -1
+		operator: %s
+		resolver: {
+			type: cache
+			dir: %s
+		}
+		leafnodes: {
+			%s
+		}
+	`, ojwt, tmpDir, cfg)))
+			defer os.Remove(conf)
+			opts := LoadConfig(conf)
+			s, err := NewServer(opts)
+			if err == nil {
+				s.Shutdown()
+				t.Fatal("Expected an error")
+			}
+			if err.Error() != errorText {
+				t.Fatalf("Expected error %s but got %s", errorText, err)
+			}
+		})
+	}
+}
diff --git a/test/jetstream_test.go b/test/jetstream_test.go
index 70143011..f9090d7d 100644
--- a/test/jetstream_test.go
+++ b/test/jetstream_test.go
@@ -285,8 +285,8 @@ func TestJetStreamAddStreamDiscardNew(t *testing.T) {
 			if resp == nil {
 				t.Fatalf("No response, possible timeout?")
 			}
-			if string(resp.Data) != "-ERR 'maximum messages exceeded'" {
-				t.Fatalf("Expected to get an error about maximum messages, got %q", resp.Data)
+			if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error.Description != "maximum messages exceeded" {
+				t.Fatalf("Expected to get an error about maximum messages, got %q", pa.Error)
 			}
 
 			// Now do bytes.
@@ -297,7 +297,7 @@ func TestJetStreamAddStreamDiscardNew(t *testing.T) {
 			if resp == nil {
 				t.Fatalf("No response, possible timeout?")
 			}
-			if string(resp.Data) != "-ERR 'maximum bytes exceeded'" {
+			if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error.Description != "maximum bytes exceeded" {
 				t.Fatalf("Expected to get an error about maximum bytes, got %q", resp.Data)
 			}
 		})
@@ -382,18 +382,15 @@ func TestJetStreamPubAck(t *testing.T) {
 		if resp == nil {
 			t.Fatalf("No response from send stream msg")
 		}
-		if !bytes.HasPrefix(resp.Data, []byte("+OK {")) {
-			t.Fatalf("Did not get a correct response: %q", resp.Data)
+		pa := getPubAckResponse(resp.Data)
+		if pa == nil || pa.Error != nil {
+			t.Fatalf("Expected a valid JetStreamPubAck, got %q", resp.Data)
 		}
-		var pubAck server.PubAck
-		if err := json.Unmarshal(resp.Data[3:], &pubAck); err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-		if pubAck.Stream != sname {
-			t.Fatalf("Expected %q for stream name, got %q", sname, pubAck.Stream)
+		if pa.Stream != sname {
+			t.Fatalf("Expected %q for stream name, got %q", sname, pa.Stream)
 		}
-		if pubAck.Seq != seq {
-			t.Fatalf("Expected %d for sequence, got %d", seq, pubAck.Seq)
+		if pa.Sequence != seq {
+			t.Fatalf("Expected %d for sequence, got %d", seq, pa.Sequence)
 		}
 	}
 
@@ -914,8 +911,8 @@ func TestJetStreamAddStreamMaxMsgSize(t *testing.T) {
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
-			if string(resp.Data) != "-ERR 'message size exceeds maximum allowed'" {
-				t.Fatalf("Expected to get an error for maximum message size, got %q", resp.Data)
+			if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error.Description != "message size exceeds maximum allowed" {
+				t.Fatalf("Expected to get an error for maximum message size, got %q", pa.Error)
 			}
 		})
 	}
@@ -1105,14 +1102,11 @@ func sendStreamMsg(t *testing.T, nc *nats.Conn, subject, msg string) *server.Pub
 	if resp == nil {
 		t.Fatalf("No response for %q, possible timeout?", msg)
 	}
-	if !bytes.HasPrefix(resp.Data, []byte("+OK {")) {
-		t.Fatalf("Expected a JetStreamPubAck, got %q", resp.Data)
-	}
-	var pubAck server.PubAck
-	if err := json.Unmarshal(resp.Data[3:], &pubAck); err != nil {
-		t.Fatalf("Unexpected error: %v", err)
+	pa := getPubAckResponse(resp.Data)
+	if pa == nil || pa.Error != nil {
+		t.Fatalf("Expected a valid JetStreamPubAck, got %q", resp.Data)
 	}
-	return &pubAck
+	return pa.PubAck
 }
 
 func TestJetStreamBasicAckPublish(t *testing.T) {
@@ -3344,23 +3338,20 @@ func TestJetStreamPublishDeDupe(t *testing.T) {
 	sendMsg := func(seq uint64, id, msg string) *server.PubAck {
 		t.Helper()
 		m := nats.NewMsg(fmt.Sprintf("foo.%d", seq))
-		m.Header.Add(server.JSPubId, id)
+		m.Header.Add(server.JSMsgId, id)
 		m.Data = []byte(msg)
 		resp, _ := nc.RequestMsg(m, 100*time.Millisecond)
 		if resp == nil {
 			t.Fatalf("No response for %q, possible timeout?", msg)
 		}
-		if !bytes.HasPrefix(resp.Data, []byte("+OK {")) {
+		pa := getPubAckResponse(resp.Data)
+		if pa == nil || pa.Error != nil {
 			t.Fatalf("Expected a JetStreamPubAck, got %q", resp.Data)
 		}
-		var pubAck server.PubAck
-		if err := json.Unmarshal(resp.Data[3:], &pubAck); err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-		if pubAck.Seq != seq {
-			t.Fatalf("Did not get correct sequence in PubAck, expected %d, got %d", seq, pubAck.Seq)
+		if pa.Sequence != seq {
+			t.Fatalf("Did not get correct sequence in PubAck, expected %d, got %d", seq, pa.Sequence)
 		}
-		return &pubAck
+		return pa.PubAck
 	}
 
 	expect := func(n uint64) {
@@ -3415,7 +3406,6 @@ func TestJetStreamPublishDeDupe(t *testing.T) {
 	if err := mset.Update(&cfg); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-
 	mset.Purge()
 
 	// Send 5 new messages.
@@ -3465,6 +3455,110 @@ func TestJetStreamPublishDeDupe(t *testing.T) {
 	nmids(0)
 }
 
+func getPubAckResponse(msg []byte) *server.JSPubAckResponse {
+	var par server.JSPubAckResponse
+	if err := json.Unmarshal(msg, &par); err != nil {
+		return nil
+	}
+	return &par
+}
+
+func TestJetStreamPublishExpect(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer os.RemoveAll(config.StoreDir)
+	}
+
+	mname := "EXPECT"
+	mset, err := s.GlobalAccount().AddStream(&server.StreamConfig{Name: mname, Storage: server.FileStorage, MaxAge: time.Hour, Subjects: []string{"foo.*"}})
+	if err != nil {
+		t.Fatalf("Unexpected error adding stream: %v", err)
+	}
+	defer mset.Delete()
+
+	nc := clientConnectToServer(t, s)
+	defer nc.Close()
+
+	// Test that we get no error when expected stream is correct.
+	m := nats.NewMsg("foo.bar")
+	m.Data = []byte("HELLO")
+	m.Header.Set(server.JSExpectedStream, mname)
+	resp, err := nc.RequestMsg(m, 100*time.Millisecond)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error != nil {
+		t.Fatalf("Expected a valid JetStreamPubAck, got %q", resp.Data)
+	}
+
+	// Now test that we get an error back when expecting a different stream.
+	m.Header.Set(server.JSExpectedStream, "ORDERS")
+	resp, err = nc.RequestMsg(m, 100*time.Millisecond)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error == nil {
+		t.Fatalf("Expected an error, got %q", resp.Data)
+	}
+
+	// Now test that we get an error back when expecting a different sequence number.
+	m.Header.Set(server.JSExpectedStream, mname)
+	m.Header.Set(server.JSExpectedLastSeq, "10")
+	resp, err = nc.RequestMsg(m, 100*time.Millisecond)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error == nil {
+		t.Fatalf("Expected an error, got %q", resp.Data)
+	}
+
+	// Now send a message with a message ID and make sure we can match that.
+	m = nats.NewMsg("foo.bar")
+	m.Data = []byte("HELLO")
+	m.Header.Set(server.JSMsgId, "AAA")
+	if _, err = nc.RequestMsg(m, 100*time.Millisecond); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Now try again with new message ID but require last one to be 'BBB'
+	m.Header.Set(server.JSMsgId, "ZZZ")
+	m.Header.Set(server.JSExpectedLastMsgId, "BBB")
+	resp, err = nc.RequestMsg(m, 100*time.Millisecond)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error == nil {
+		t.Fatalf("Expected an error, got %q", resp.Data)
+	}
+
+	// Restart the server and make sure we remember/rebuild last seq and last msgId.
+	// Stop current server.
+	sd := s.JetStreamConfig().StoreDir
+	s.Shutdown()
+	// Restart.
+	s = RunJetStreamServerOnPort(-1, sd)
+	defer s.Shutdown()
+
+	nc = clientConnectToServer(t, s)
+	defer nc.Close()
+
+	// Our last sequence was 2 and last msgId was "AAA"
+	m = nats.NewMsg("foo.baz")
+	m.Data = []byte("HELLO AGAIN")
+	m.Header.Set(server.JSExpectedLastSeq, "2")
+	m.Header.Set(server.JSExpectedLastMsgId, "AAA")
+	m.Header.Set(server.JSMsgId, "BBB")
+	resp, err = nc.RequestMsg(m, 100*time.Millisecond)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error != nil {
+		t.Fatalf("Expected a valid JetStreamPubAck, got %q", resp.Data)
+	}
+}
+
 func TestJetStreamPullConsumerRemoveInterest(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	defer s.Shutdown()
@@ -3852,6 +3946,15 @@ func TestJetStreamConsumerMaxDeliveryAndServerRestart(t *testing.T) {
 		s = RunJetStreamServerOnPort(port, sd)
 	}
 
+	waitForClientReconnect := func() {
+		checkFor(t, 2500*time.Millisecond, 5*time.Millisecond, func() error {
+			if !nc.IsConnected() {
+				return fmt.Errorf("Not connected")
+			}
+			return nil
+		})
+	}
+
 	// Restart.
 	restartServer()
 	defer s.Shutdown()
@@ -3859,12 +3962,7 @@ func TestJetStreamConsumerMaxDeliveryAndServerRestart(t *testing.T) {
 	checkNumMsgs(2)
 
 	// Wait for client to be reconnected.
-	checkFor(t, 2500*time.Millisecond, 5*time.Millisecond, func() error {
-		if !nc.IsConnected() {
-			return fmt.Errorf("Not connected")
-		}
-		return nil
-	})
+	waitForClientReconnect()
 
 	// Once we are here send third order.
 	sendStreamMsg(t, nc, mname, "order-3")
@@ -3876,6 +3974,9 @@ func TestJetStreamConsumerMaxDeliveryAndServerRestart(t *testing.T) {
 
 	checkNumMsgs(3)
 
+	// Wait for client to be reconnected.
+	waitForClientReconnect()
+
 	// Now we should have max times three on our sub.
 	checkSubPending(max * 3)
 
@@ -8756,6 +8857,46 @@ func TestJetStreamPubPerf(t *testing.T) {
 	fmt.Printf("%.0f msgs/sec\n", float64(toSend)/tt.Seconds())
 }
 
+func TestJetStreamPubWithAsyncResponsePerf(t *testing.T) {
+	// Comment out to run, holding place for now.
+	t.SkipNow()
+
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer os.RemoveAll(config.StoreDir)
+	}
+
+	acc := s.GlobalAccount()
+
+	msetConfig := server.StreamConfig{
+		Name:     "sr33",
+		Storage:  server.MemoryStorage,
+		Subjects: []string{"foo"},
+	}
+
+	if _, err := acc.AddStream(&msetConfig); err != nil {
+		t.Fatalf("Unexpected error adding stream: %v", err)
+	}
+
+	nc := clientConnectToServer(t, s)
+	defer nc.Close()
+
+	toSend := 1000000
+	payload := []byte("Hello World")
+
+	start := time.Now()
+	for i := 0; i < toSend; i++ {
+		nc.PublishRequest("foo", "bar", payload)
+	}
+	nc.Flush()
+
+	tt := time.Since(start)
+	fmt.Printf("time is %v\n", tt)
+	fmt.Printf("%.0f msgs/sec\n", float64(toSend)/tt.Seconds())
+}
+
 func TestJetStreamConsumerPerf(t *testing.T) {
 	// Comment out to run, holding place for now.
 	t.SkipNow()
@@ -9860,7 +10001,7 @@ func TestJetStreamAccountImportBasics(t *testing.T) {
 
 	// Simple publish to a stream.
 	pubAck := sendStreamMsg(t, nc, "my.orders.foo", "ORDERS-1")
-	if pubAck.Stream != "ORDERS" || pubAck.Seq != 1 {
+	if pubAck.Stream != "ORDERS" || pubAck.Sequence != 1 {
 		t.Fatalf("Bad pubAck received: %+v", pubAck)
 	}
 	if msgs := mset.State().Msgs; msgs != 1 {
diff --git a/test/service_latency_test.go b/test/service_latency_test.go
index 00287a20..5ebd57c9 100644
--- a/test/service_latency_test.go
+++ b/test/service_latency_test.go
@@ -1303,6 +1303,66 @@ func TestServiceAndStreamStackOverflow(t *testing.T) {
 	})
 }
 
+func TestServiceCycle(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		accounts {
+		  A {
+		    exports [ { service: help } ]
+			imports [ { service { subject: help, account: B } } ]
+		  }
+		  B {
+		    exports [ { service: help } ]
+			imports [ { service { subject: help, account: A } } ]
+		  }
+		}
+	`))
+	defer os.Remove(conf)
+
+	if _, err := server.ProcessConfigFile(conf); err == nil || !strings.Contains(err.Error(), server.ErrServiceImportFormsCycle.Error()) {
+		t.Fatalf("Expected an error on cycle service import, got none")
+	}
+
+	conf = createConfFile(t, []byte(`
+		accounts {
+		  A {
+		    exports [ { service: * } ]
+			imports [ { service { subject: help, account: B } } ]
+		  }
+		  B {
+		    exports [ { service: help } ]
+			imports [ { service { subject: *, account: A } } ]
+		  }
+		}
+	`))
+	defer os.Remove(conf)
+
+	if _, err := server.ProcessConfigFile(conf); err == nil || !strings.Contains(err.Error(), server.ErrServiceImportFormsCycle.Error()) {
+		t.Fatalf("Expected an error on cycle service import, got none")
+	}
+
+	conf = createConfFile(t, []byte(`
+		accounts {
+		  A {
+		    exports [ { service: * } ]
+			imports [ { service { subject: help, account: B } } ]
+		  }
+		  B {
+		    exports [ { service: help } ]
+			imports [ { service { subject: help, account: C } } ]
+		  }
+		  C {
+		    exports [ { service: * } ]
+			imports [ { service { subject: *, account: A } } ]
+		  }
+		}
+	`))
+	defer os.Remove(conf)
+
+	if _, err := server.ProcessConfigFile(conf); err == nil || !strings.Contains(err.Error(), server.ErrServiceImportFormsCycle.Error()) {
+		t.Fatalf("Expected an error on cycle service import, got none")
+	}
+}
+
 // Check we get the proper detailed information for the requestor when allowed.
 func TestServiceLatencyRequestorSharesDetailedInfo(t *testing.T) {
 	sc := createSuperCluster(t, 3, 3)