[FIXED] Nil stream assignment during processClusterUpdateStream scale-up (nats-io#8049)

When `processClusterUpdateStream` takes the scale-up branch it calls
`mset.sa.Sync` in `startClusterSubs`. If the stream has just been
recovered and meta catchup hasn't yet assigned the stream (`mset.sa ==
nil`), this panics with a nil pointer deref.

Resolves nats-io#7229

Signed-off-by: Maurice van Veen <github@mauricevanveen.com>

Author: neilalexander

server/jetstream_cluster.go

@@ -4933,6 +4933,9 @@ func (js *jetStream) processClusterUpdateStream(acc *Account, osa, sa *streamAss
 
 		if !alreadyRunning && numReplicas > 1 {
 			if needsNode {
+				// Must run before startClusterSubs reads mset.sa.Sync.
+				mset.setStreamAssignment(sa)
+
 				// Since we are scaling up we want to make sure our sync subject
 				// is registered before we start our raft node.
 				mset.mu.Lock()

server/jetstream_cluster_1_test.go

@@ -6785,6 +6785,72 @@ func TestJetStreamClusterMetaRecoveryAddAndUpdateStream(t *testing.T) {
 	require_Len(t, len(sa.Config.Subjects), 1)
 }
 
+// https://github.com/nats-io/nats-server/issues/7229
+func TestJetStreamClusterProcessClusterUpdateStreamNilStreamAssignment(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "TEST", Subjects: []string{"foo"}, Replicas: 3})
+	require_NoError(t, err)
+
+	s := c.randomServer()
+	acc, err := s.lookupAccount(globalAccountName)
+	require_NoError(t, err)
+	mset, err := acc.lookupStream("TEST")
+	require_NoError(t, err)
+
+	// Capture osa and build a sa that reflects the post-restart meta-replay
+	// conditions: a scale-up path where both the old and new raftGroup nodes
+	// are nil on this server (alreadyRunning=false, needsNode=true).
+	sjs := s.getJetStream()
+	sjs.mu.Lock()
+	realOsa := sjs.streamAssignment(globalAccountName, "TEST")
+	if realOsa == nil {
+		sjs.mu.Unlock()
+		t.Fatal("stream assignment not found")
+	}
+
+	// Intentionally omit Group.node so that in processClusterUpdateStream
+	// alreadyRunning (osa.Group.node != nil) is false and needsNode
+	// (sa.Group.node == nil) is true, taking the scale-up branch.
+	cloneGroup := func() *raftGroup {
+		return &raftGroup{
+			Name:    realOsa.Group.Name,
+			Peers:   append([]string{}, realOsa.Group.Peers...),
+			Storage: realOsa.Group.Storage,
+		}
+	}
+	osa := *realOsa
+	osa.Group = cloneGroup()
+	sa := osa
+	cfg := *realOsa.Config
+	cfg.MaxMsgs = 1_000
+	sa.Config = &cfg
+	sa.Group = cloneGroup()
+	sjs.mu.Unlock()
+
+	// Simulate the state right after recoverStream() before meta catchup:
+	// no stream assignment, no raft node, no cluster subs registered yet.
+	mset.mu.Lock()
+	mset.sa = nil
+	mset.node = nil
+	if mset.syncSub != nil {
+		mset.srv.sysUnsubscribe(mset.syncSub)
+		mset.syncSub = nil
+	}
+	mset.mu.Unlock()
+
+	sjs.processClusterUpdateStream(acc, &osa, &sa)
+
+	mset.mu.RLock()
+	got := mset.sa
+	mset.mu.RUnlock()
+	require_NotNil(t, got)
+}
+
 // Make sure if we received acks that are out of bounds, meaning past our
 // last sequence or before our first that they are ignored and errored if applicable.
 func TestJetStreamClusterConsumerAckOutOfBounds(t *testing.T) {