Adopt orphaned R1 streams on standalone-to-cluster promotion

Signed-off-by: Tilak Raj <tilak.raj94@gmail.com>

Author: tilakraj94

server/jetstream_cluster.go

@@ -80,6 +80,13 @@ type jetStreamCluster struct {
 	// Track last meta snapshot time and duration for monitoring.
 	lastMetaSnapTime     int64 // Unix nanoseconds
 	lastMetaSnapDuration int64 // Duration in nanoseconds
+	// Signals that this node was promoted from standalone to clustered mode
+	// with no prior meta state. Used to detect standalone-to-clustered transitions.
+	ptc bool
+	// Tracks streams that were preserved (not deleted) due to name conflicts
+	// with existing cluster assignments during ptc promotion. This prevents
+	// the ptc marker from being cleaned up while unresolved conflicts remain.
+	ptcConflicts int
 }
 
 // Used to track inflight stream create/update/delete requests that have been proposed but not yet applied.
@@ -362,6 +369,13 @@ const (
 	defaultMetaGroupName = "_meta_"
 	defaultMetaFSBlkSize = 1024 * 1024
 	jsExcludePlacement   = "!jetstream"
+
+	// ptcMarkerFile is a sentinel file written into the meta Raft group directory
+	// when a standalone server is promoted to a cluster member and has R1 streams
+	// pending adoption. Its presence tells checkForOrphans (and processStreamAssignment)
+	// that local R1 stream data must be preserved, not deleted, until all orphans
+	// have been successfully adopted.
+	ptcMarkerFile = "ptc"
 )
 
 // Returns information useful in mixed mode.
@@ -870,6 +884,51 @@ func (s *Server) enableJetStreamClustering() error {
 	return js.setupMetaGroup()
 }
 
+func hasPTCMarker(storeDir string) bool {
+	_, err := os.Stat(filepath.Join(storeDir, ptcMarkerFile))
+	return err == nil
+}
+
+func (js *jetStream) ptcMarkerPath() string {
+	s := js.srv
+	sysAcc := s.SystemAccount()
+	if sysAcc == nil {
+		return _EMPTY_
+	}
+	return filepath.Join(js.config.StoreDir, sysAcc.Name, defaultStoreDirName, defaultMetaGroupName, ptcMarkerFile)
+}
+
+func (js *jetStream) writePTCMarker() error {
+	p := js.ptcMarkerPath()
+	if p != _EMPTY_ {
+		return os.WriteFile(p, []byte{}, defaultFilePerms)
+	}
+
+	return nil
+}
+
+func (js *jetStream) removePTCMarker() error {
+	var err error
+	p := js.ptcMarkerPath()
+	if p != _EMPTY_ {
+		err = os.Remove(p)
+		if errors.Is(err, os.ErrNotExist) {
+			return nil
+		}
+	}
+
+	return err
+}
+
+func (js *jetStream) hasPTCMarker() bool {
+	p := js.ptcMarkerPath()
+	if p == _EMPTY_ {
+		return false
+	}
+	_, err := os.Stat(p)
+	return err == nil
+}
+
 // isClustered returns if we are clustered.
 // Lock should not be held.
 func (js *jetStream) isClustered() bool {
@@ -972,6 +1031,11 @@ func (js *jetStream) setupMetaGroup() error {
 	// If we are bootstrapped with no state, start campaign early.
 	if bootstrap {
 		n.Campaign()
+
+		// Write the PTC marker immediately on bootstrap
+		if err = os.WriteFile(filepath.Join(storeDir, ptcMarkerFile), []byte{}, defaultFilePerms); err != nil {
+			s.Fatalf("Failed to write PTC marker on bootstrap: %v", err)
+		}
 	}
 
 	c := s.createInternalJetStreamClient()
@@ -985,6 +1049,7 @@ func (js *jetStream) setupMetaGroup() error {
 		c:       c,
 		qch:     make(chan struct{}),
 		stopped: make(chan struct{}),
+		ptc:     bootstrap || hasPTCMarker(storeDir),
 	}
 	atomic.StoreInt32(&js.clustered, 1)
 	c.registerWithAccount(sysAcc)
@@ -1390,50 +1455,190 @@ func (js *jetStream) checkForOrphans() {
 	s, cc := js.srv, js.cluster
 	s.Debugf("JetStream cluster checking for orphans")
 
-	// We only want to cleanup any orphans if we know we are current with the meta-leader.
+	// Skip if there is no leader or if this node has not yet applied all committed Raft entries;
 	meta := cc.meta
 	if meta == nil || meta.Leaderless() {
 		js.mu.Unlock()
-		s.Debugf("JetStream cluster skipping check for orphans, no meta-leader")
+		s.Debugf("JetStream cluster skipping orphan check, no meta-leader")
 		return
 	}
-	if !meta.Healthy() {
+	if !meta.Current() {
 		js.mu.Unlock()
-		s.Debugf("JetStream cluster skipping check for orphans, not current with the meta-leader")
+		s.Debugf("JetStream cluster skipping orphan check, meta not current yet will retry")
+		time.AfterFunc(5*time.Second, js.checkForOrphans)
 		return
 	}
+	ourPeerID := meta.ID()
 	streams, consumers := js.getOrphans()
+	ptc := cc.ptc
+	// the counter tells us here that they still need operator attention so we must
+	// keep the PTC marker alive.
+	ptcConflicts := cc.ptcConflicts
 	js.mu.Unlock()
 
+	// hasUnresolved tracks whether any R1 stream still needs attention after this
+	// run. It is used to decide whether the PTC marker can be cleaned up.
+	hasUnresolved := ptcConflicts > 0
+
 	for _, mset := range streams {
 		mset.mu.RLock()
-		accName, stream := mset.acc.Name, mset.cfg.Name
+		accName := mset.acc.Name
+		cfg := mset.cfg
+		created := mset.created
 		mset.mu.RUnlock()
-		s.Warnf("Detected orphaned stream '%s > %s', will cleanup", accName, stream)
+
+		// R1 streams on a PTC node are adopted into the cluster instead of deleted.
+		// Any other orphan (R>1, or R1 without the PTC flag) is cleaned up normally.
+		if cfg.Replicas == 1 && ptc {
+			cfgCopy := cfg
+			sa := &streamAssignment{
+				Client:  &ClientInfo{Account: accName},
+				Config:  &cfgCopy,
+				Created: created,
+				Sync:    syncSubjForStream(),
+				Group: &raftGroup{
+					Name:    groupNameForStream([]string{ourPeerID}, cfg.Storage),
+					Peers:   []string{ourPeerID},
+					Storage: cfg.Storage,
+				},
+			}
+			sa.ConfigJSON, _ = json.Marshal(cfgCopy)
+			s.Noticef("Adopting orphaned R1 stream '%s > %s' into cluster assignments", accName, cfg.Name)
+			if err := meta.ForwardProposal(encodeAddStreamAssignment(sa)); err != nil {
+				s.Warnf("Failed to propose adoption of R1 stream '%s > %s': %v", accName, cfg.Name, err)
+				hasUnresolved = true
+				continue
+			}
+			// Proposal was sent but Raft commit is async. Keep the marker so that a
+			// crash between here and the commit does not lose the resume signal. The
+			// next sweep will find no orphans once the entry is committed and will
+			// then remove the marker cleanly.
+			hasUnresolved = true
+
+			// Propose assignments for durable consumers on this stream.
+			// Ephemeral consumers are transient; they do not need adoption.
+			for _, o := range mset.getConsumers() {
+				if !o.isDurable() {
+					continue
+				}
+				oCfg := o.config()
+				oCreated := o.createdTime()
+				oStorage := cfg.Storage
+				if oCfg.MemoryStorage {
+					oStorage = MemoryStorage
+				}
+				ca := &consumerAssignment{
+					Client:  &ClientInfo{Account: accName},
+					Name:    o.String(),
+					Stream:  cfg.Name,
+					Config:  &oCfg,
+					Created: oCreated,
+					Group: &raftGroup{
+						Name:    groupNameForConsumer([]string{ourPeerID}, oStorage),
+						Peers:   []string{ourPeerID},
+						Storage: oStorage,
+					},
+				}
+				ca.ConfigJSON, _ = json.Marshal(oCfg)
+				s.Noticef("Adopting orphaned consumer '%s > %s > %s' into cluster assignments", accName, cfg.Name, o.String())
+				if err := meta.ForwardProposal(encodeAddConsumerAssignment(ca)); err != nil {
+					s.Warnf("Failed to propose adoption of consumer '%s > %s > %s': %v", accName, cfg.Name, o.String(), err)
+				}
+				// Consumer proposal is also async; hasUnresolved is already true from
+				// the stream proposal above, so no additional set is needed here.
+			}
+			continue
+		}
+
+		// Orphaned stream with no valid adoption path: clean up locally.
+		s.Warnf("Detected orphaned stream '%s > %s', will cleanup", accName, cfg.Name)
 		if err := mset.delete(); err != nil {
 			s.Warnf("Deleting stream encountered an error: %v", err)
 		}
 	}
+
 	for _, o := range consumers {
 		o.mu.RLock()
-		accName, mset, consumer := o.acc.Name, o.mset, o.name
+		accName, mset, oName := o.acc.Name, o.mset, o.name
 		o.mu.RUnlock()
-		stream := "N/A"
+
+		streamName := "N/A"
+		var streamCfg StreamConfig
 		if mset != nil {
 			mset.mu.RLock()
-			stream = mset.cfg.Name
+			streamName = mset.cfg.Name
+			streamCfg = mset.cfg
 			mset.mu.RUnlock()
 		}
+
+		// If the parent stream is R1 and this is a PTC node, adopt the durable consumer
+		if streamCfg.Replicas == 1 && o.isDurable() && ptc {
+			oCfg := o.config()
+			oCreated := o.createdTime()
+			oStorage := streamCfg.Storage
+			if oCfg.MemoryStorage {
+				oStorage = MemoryStorage
+			}
+			ca := &consumerAssignment{
+				Client:  &ClientInfo{Account: accName},
+				Name:    oName,
+				Stream:  streamName,
+				Config:  &oCfg,
+				Created: oCreated,
+				Group: &raftGroup{
+					Name:    groupNameForConsumer([]string{ourPeerID}, oStorage),
+					Peers:   []string{ourPeerID},
+					Storage: oStorage,
+				},
+			}
+			ca.ConfigJSON, _ = json.Marshal(oCfg)
+			s.Noticef("Adopting orphaned consumer '%s > %s > %s' into cluster assignments", accName, streamName, oName)
+			if err := meta.ForwardProposal(encodeAddConsumerAssignment(ca)); err != nil {
+				s.Warnf("Failed to propose adoption of consumer '%s > %s > %s': %v", accName, streamName, oName, err)
+			}
+			// Proposal is async; keep the marker until the next sweep confirms the
+			// consumer is no longer an orphan after the Raft entry is committed.
+			hasUnresolved = true
+			continue
+		}
+
 		if o.isDurable() {
-			s.Warnf("Detected orphaned durable consumer '%s > %s > %s', will cleanup", accName, stream, consumer)
+			s.Warnf("Detected orphaned durable consumer '%s > %s > %s', will cleanup", accName, streamName, oName)
 		} else {
-			s.Debugf("Detected orphaned consumer '%s > %s > %s', will cleanup", accName, stream, consumer)
+			s.Debugf("Detected orphaned consumer '%s > %s > %s', will cleanup", accName, streamName, oName)
 		}
-
 		if err := o.delete(); err != nil {
 			s.Warnf("Deleting consumer encountered an error: %v", err)
 		}
 	}
+
+	// PTC marker lifecycle:
+	//   - Keep it if there are still unresolved conflicts or remaining orphans
+	//     (they will be retried / resolved on the next orphan sweep).
+	//   - Remove it once everything has been adopted and no conflicts remain.
+	if !ptc {
+		return
+	}
+
+	if hasUnresolved {
+		err := js.writePTCMarker()
+		if err != nil {
+			s.Warnf("Failed to write ptc marker: %v", err)
+		}
+	} else {
+		if err := js.removePTCMarker(); err != nil {
+			s.Warnf("Failed to remove ptc marker: %v", err)
+		}
+
+		// Clear the in-memory flag so future orphan sweeps don't re-enter
+		// the PTC adoption path unnecessarily.
+		js.mu.Lock()
+		if js.cluster != nil {
+			js.cluster.ptc = false
+			js.cluster.ptcConflicts = 0
+		}
+		js.mu.Unlock()
+	}
 }
 
 // Returns orphaned streams and consumers that were recovered from disk, but don't
@@ -2025,11 +2230,34 @@ func (js *jetStream) applyMetaSnapshot(buf []byte, ru *recoveryUpdates, isRecove
 		// logging and sending out advisories.
 		js.mu.RLock()
 		deleteStreams, deleteConsumers := js.getOrphans()
+		cc := js.cluster
+		ptc := cc != nil && cc.ptc
 		js.mu.RUnlock()
 		for _, mset := range deleteStreams {
+			mset.mu.RLock()
+			cfg := mset.cfg
+			mset.mu.RUnlock()
+			// During active PTC promotion, preserve R1 orphans: the meta-leader's snapshot
+			// may not yet include the adoption proposal. checkForOrphans will handle them.
+			if cfg.Replicas == 1 && ptc {
+				continue
+			}
 			mset.stop(true, false)
 		}
 		for _, o := range deleteConsumers {
+			o.mu.RLock()
+			mset := o.mset
+			o.mu.RUnlock()
+			if mset != nil {
+				mset.mu.RLock()
+				replicas := mset.cfg.Replicas
+				mset.mu.RUnlock()
+				// During active PTC promotion, preserve consumers whose parent stream is R1:
+				// the consumer adoption proposal may still be in-flight. checkForOrphans will handle them.
+				if replicas == 1 && ptc {
+					continue
+				}
+			}
 			o.deleteWithoutAdvisory()
 		}
 	}
@@ -4565,6 +4793,7 @@ func (js *jetStream) processStreamAssignment(sa *streamAssignment) {
 	if sa.Group != nil && ourID != _EMPTY_ {
 		isMember = sa.Group.isMember(ourID)
 	}
+	ptc := !noMeta && cc.ptc
 
 	if s == nil || noMeta {
 		js.mu.Unlock()
@@ -4648,8 +4877,23 @@ func (js *jetStream) processStreamAssignment(sa *streamAssignment) {
 	if isMember {
 		js.processClusterCreateStream(acc, sa)
 	} else if mset, _ := acc.lookupStream(sa.Config.Name); mset != nil {
-		// We have one here even though we are not a member. This can happen on re-assignment.
-		s.removeStream(mset, sa)
+		mset.mu.RLock()
+		replicas := mset.cfg.Replicas
+		mset.mu.RUnlock()
+		if ptc && replicas == 1 {
+			// On a node promoted from standalone to clustered, do not delete
+			// conflicting R1 streams. Stop the stream to free resources but preserve
+			// data on disk so the operator can recover it (e.g. rename the stream, and re-promote).
+			s.Warnf("JetStream preserving local stream '%s > %s' data on disk: conflicts with existing cluster assignment",
+				accName, sa.Config.Name)
+			mset.stop(false, false)
+			js.mu.Lock()
+			cc.ptcConflicts++
+			js.mu.Unlock()
+		} else {
+			// We have one here even though we are not a member. This can happen on re-assignment.
+			s.removeStream(mset, sa)
+		}
 	}
 
 	// If this stream assignment does not have a sync subject (bug) set that the meta-leader should check when elected.
@@ -5388,6 +5632,7 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 	if ca.Group != nil && ourID != _EMPTY_ {
 		isMember = ca.Group.isMember(ourID)
 	}
+	ptc := !noMeta && cc.ptc
 	js.mu.RUnlock()
 
 	if s == nil || noMeta || shuttingDown {
@@ -5495,8 +5740,17 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 		js.processClusterCreateConsumer(oca, ca, state, wasExisting)
 	} else if mset, _ := acc.lookupStream(sa.Config.Name); mset != nil {
 		if o := mset.lookupConsumer(ca.Name); o != nil {
-			// We have one here even though we are not a member. This can happen on re-assignment.
-			s.removeConsumer(o, ca)
+			mset.mu.RLock()
+			replicas := mset.cfg.Replicas
+			mset.mu.RUnlock()
+			if ptc && replicas == 1 {
+				// On a node promoted from standalone to clustered, preserve local consumers
+				// of R1 streams. Data is on disk; checkForOrphans will adopt them.
+				s.Debugf("JetStream preserving local consumer '%s > %s > %s' during PTC promotion", accName, stream, ca.Name)
+			} else {
+				// We have one here even though we are not a member. This can happen on re-assignment.
+				s.removeConsumer(o, ca)
+			}
 		}
 	}
 }