Cherry-pick de-flakes for 2.10.25-RC.1 (#6349)

Deponds on #6345

Includes the following de-flakes:
- #6329
- #6330
- #6331
- #6332
- #6334

And this data race fix:
- #6150

Signed-off-by: Maurice van Veen <[email protected]>

Author: neilalexander

server/jetstream_cluster_2_test.go

@@ -4219,7 +4219,7 @@ func TestJetStreamClusterRedeliverBackoffs(t *testing.T) {
 		d := tr.Sub(start)
 		// Adjust start for next calcs.
 		start = start.Add(d)
-		if d < expected[i] || d > expected[i]*2 {
+		if d < expected[i]-5*time.Millisecond || d > expected[i]*2+5*time.Millisecond {
 			t.Fatalf("Timing is off for %d, expected ~%v, but got %v", i, expected[i], d)
 		}
 	}
@@ -6238,8 +6238,12 @@ func TestJetStreamClusterStreamResetOnExpirationDuringPeerDownAndRestartWithLead
 
 	// Wait for all messages to expire.
 	checkFor(t, 5*time.Second, time.Second, func() error {
-		si, err := js.StreamInfo("TEST")
-		require_NoError(t, err)
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		si, err := js.StreamInfo("TEST", nats.Context(ctx))
+		if err != nil {
+			return err
+		}
 		if si.State.Msgs == 0 {
 			return nil
 		}
@@ -6277,9 +6281,11 @@ func TestJetStreamClusterStreamResetOnExpirationDuringPeerDownAndRestartWithLead
 	}
 
 	// Now move the leader there and double check, but above test is sufficient.
-	checkFor(t, 10*time.Second, 250*time.Millisecond, func() error {
+	checkFor(t, 30*time.Second, 250*time.Millisecond, func() error {
 		_, err = nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, "TEST"), nil, time.Second)
-		require_NoError(t, err)
+		if err != nil {
+			return err
+		}
 		c.waitOnStreamLeader("$G", "TEST")
 		if c.streamLeader("$G", "TEST") == nsl {
 			return nil

server/jetstream_cluster_4_test.go

@@ -3603,7 +3603,9 @@ func TestJetStreamClusterDesyncAfterErrorDuringCatchup(t *testing.T) {
 				for _, n := range server.raftNodes {
 					rn := n.(*raft)
 					if rn.accName == "$G" {
+						rn.Lock()
 						rn.updateLeader(noLeader)
+						rn.Unlock()
 					}
 				}
 

server/jetstream_helpers_test.go

@@ -556,6 +556,27 @@ func (sc *supercluster) waitOnLeader() {
 	sc.t.Fatalf("Expected a cluster leader, got none")
 }
 
+func (sc *supercluster) waitOnAccount(account string) {
+	sc.t.Helper()
+	expires := time.Now().Add(40 * time.Second)
+	for time.Now().Before(expires) {
+		found := true
+		for _, c := range sc.clusters {
+			for _, s := range c.servers {
+				acc, err := s.fetchAccount(account)
+				found = found && err == nil && acc != nil
+			}
+		}
+		if found {
+			return
+		}
+		time.Sleep(100 * time.Millisecond)
+		continue
+	}
+
+	sc.t.Fatalf("Expected account %q to exist but didn't", account)
+}
+
 func (sc *supercluster) waitOnAllCurrent() {
 	sc.t.Helper()
 	for _, c := range sc.clusters {

server/jetstream_jwt_test.go

@@ -325,6 +325,8 @@ func TestJetStreamJWTMove(t *testing.T) {
 		require_False(t, s.JetStreamEnabled())
 		updateJwt(t, s.ClientURL(), sysCreds, accJwt, 10)
 
+		sc.waitOnAccount(aExpPub)
+
 		s = sc.serverByName("C2-S1")
 		require_False(t, s.JetStreamEnabled())
 
@@ -609,6 +611,8 @@ func TestJetStreamJWTClusteredTiersChange(t *testing.T) {
 	updateJwt(t, c.randomServer().ClientURL(), sysCreds, sysJwt, 3)
 	updateJwt(t, c.randomServer().ClientURL(), sysCreds, accJwt1, 3)
 
+	c.waitOnAccount(aExpPub)
+
 	nc := natsConnect(t, c.randomServer().ClientURL(), nats.UserCredentials(accCreds))
 	defer nc.Close()
 
@@ -693,6 +697,8 @@ func TestJetStreamJWTClusteredDeleteTierWithStreamAndMove(t *testing.T) {
 	updateJwt(t, c.randomServer().ClientURL(), sysCreds, sysJwt, 3)
 	updateJwt(t, c.randomServer().ClientURL(), sysCreds, accJwt1, 3)
 
+	c.waitOnAccount(aExpPub)
+
 	nc := natsConnect(t, c.randomServer().ClientURL(), nats.UserCredentials(accCreds))
 	defer nc.Close()
 

server/jetstream_test.go

@@ -8536,10 +8536,14 @@ func TestJetStreamNextMsgNoInterest(t *testing.T) {
 				}
 			}
 			nc.Flush()
-			ostate := o.info()
-			if ostate.AckFloor.Stream != 11 || ostate.NumAckPending > 0 {
-				t.Fatalf("Inconsistent ack state: %+v", ostate)
-			}
+
+			checkFor(t, time.Second, 10*time.Millisecond, func() error {
+				ostate := o.info()
+				if ostate.AckFloor.Stream != 11 || ostate.NumAckPending > 0 {
+					return fmt.Errorf("Inconsistent ack state: %+v", ostate)
+				}
+				return nil
+			})
 		})
 	}
 }
@@ -23018,6 +23022,7 @@ func TestJetStreamConsumerDontDecrementPendingCountOnSkippedMsg(t *testing.T) {
 	o := mset.lookupConsumer("CONSUMER")
 
 	requireExpected := func(expected int64) {
+		t.Helper()
 		checkFor(t, time.Second, 10*time.Millisecond, func() error {
 			o.mu.RLock()
 			npc := o.npc