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,52 @@ func (s *Server) enableJetStreamClustering() error {
 	return js.setupMetaGroup()
 }
 
+// Used during setupMetaGroup before js.cluster is initialised.
+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 +1032,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 +1050,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 +1456,181 @@ 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.
+	// We only want to cleanup any orphans if we know there is a meta-leader.
+	// ForwardProposal works on any node — followers forward to the leader automatically.
 	meta := cc.meta
 	if meta == nil || meta.Leaderless() {
 		js.mu.Unlock()
 		s.Debugf("JetStream cluster skipping check for orphans, no meta-leader")
 		return
 	}
-	if !meta.Healthy() {
-		js.mu.Unlock()
-		s.Debugf("JetStream cluster skipping check for orphans, not current with the meta-leader")
-		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)
+
+		if cfg.Replicas == 1 {
+			if !ptc {
+				s.Warnf("Detected R1 orphan stream '%s > %s' without promotion marker; "+
+					"data preserved on disk — investigate or restart to retry", accName, cfg.Name)
+				hasUnresolved = true
+				continue
+			}
+			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 := cc.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
+			}
+
+			// 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 := cc.meta.ForwardProposal(encodeAddConsumerAssignment(ca)); err != nil {
+					s.Warnf("Failed to propose adoption of consumer '%s > %s > %s': %v", accName, cfg.Name, o.String(), err)
+					hasUnresolved = true
+				}
+			}
+			continue
+		}
+
+		// R>1 orphans: Raft group exists but meta assignment is gone — clean up.
+		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 := cc.meta.ForwardProposal(encodeAddConsumerAssignment(ca)); err != nil {
+				s.Warnf("Failed to propose adoption of consumer '%s > %s > %s': %v", accName, streamName, oName, err)
+				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
@@ -2027,6 +2224,14 @@ func (js *jetStream) applyMetaSnapshot(buf []byte, ru *recoveryUpdates, isRecove
 		deleteStreams, deleteConsumers := js.getOrphans()
 		js.mu.RUnlock()
 		for _, mset := range deleteStreams {
+			mset.mu.RLock()
+			cfg := mset.cfg
+			mset.mu.RUnlock()
+			// Preserve R1 orphans: the meta-leader's snapshot may not yet include the
+			// adoption proposal. checkForOrphans will handle them 30 s after startup.
+			if cfg.Replicas == 1 {
+				continue
+			}
 			mset.stop(true, false)
 		}
 		for _, o := range deleteConsumers {
@@ -4565,6 +4770,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 +4854,20 @@ 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)
+		if ptc {
+			// On a node promoted from standalone to clustered, do not delete
+			// conflicting 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.