fission-src/pkg/utils/namespace_manager.go

package utils

import (
	"context"
	"errors"
	"sort"
	"sync"
	"time"

	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	k8sInformers "k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	k8sCache "k8s.io/client-go/tools/cache"

	managerPkg "github.com/fission/fission/pkg/utils/manager"
)

type NamespaceSubscriber interface {
	Name() string
	OnNamespaceAdd(ctx context.Context, record NamespaceRecord) error
	OnNamespaceRemove(ctx context.Context, record NamespaceRecord) error
	OnNamespaceResync(ctx context.Context, record NamespaceRecord) error
}

type NamespaceSubscriberFuncs struct {
	SubscriberName string
	AddFunc        func(ctx context.Context, record NamespaceRecord) error
	RemoveFunc     func(ctx context.Context, record NamespaceRecord) error
	ResyncFunc     func(ctx context.Context, record NamespaceRecord) error
}

func (s NamespaceSubscriberFuncs) Name() string {
	return s.SubscriberName
}

func (s NamespaceSubscriberFuncs) OnNamespaceAdd(ctx context.Context, record NamespaceRecord) error {
	if s.AddFunc == nil {
		return nil
	}
	return s.AddFunc(ctx, record)
}

func (s NamespaceSubscriberFuncs) OnNamespaceRemove(ctx context.Context, record NamespaceRecord) error {
	if s.RemoveFunc == nil {
		return nil
	}
	return s.RemoveFunc(ctx, record)
}

func (s NamespaceSubscriberFuncs) OnNamespaceResync(ctx context.Context, record NamespaceRecord) error {
	if s.ResyncFunc == nil {
		return nil
	}
	return s.ResyncFunc(ctx, record)
}

type NamespaceManager interface {
	Snapshot() []string
	SnapshotRecords() []NamespaceRecord
	Summary() NamespaceManagerSummary
	Get(name string) (NamespaceRecord, bool)
	Bootstrap(namespaces []string, source NamespaceSource, observedAt time.Time) []NamespaceRecord
	BootstrapAndDispatch(ctx context.Context, namespaces []string, source NamespaceSource, observedAt time.Time) ([]NamespaceRecord, error)
	Subscribe(subscriber NamespaceSubscriber)
	SnapshotSubscribers() []string
	Upsert(event NamespaceEvent) NamespaceRecord
	DispatchAdd(ctx context.Context, namespace string) (NamespaceRecord, bool, error)
	DispatchRemove(ctx context.Context, namespace string) (NamespaceRecord, bool, error)
	DispatchResync(ctx context.Context, namespace string) (NamespaceRecord, bool, error)
	MarkPartState(namespace string, part string, state NamespacePartState) (NamespaceRecord, bool)
	MarkPartRegistering(namespace string, part string) (NamespaceRecord, bool)
	MarkPartActive(namespace string, part string) (NamespaceRecord, bool)
	MarkPartFailed(namespace string, part string, err error) (NamespaceRecord, bool)
	Remove(name string) bool
	// RunReconciler periodically retries namespaces stuck in NamespacePhaseFailed.
	// Must be started as a goroutine; exits when ctx is cancelled.
	// logger is used to report retry attempts and outcomes; pass zap.NewNop() to silence.
	RunReconciler(ctx context.Context, logger *zap.Logger)
}

type inMemoryNamespaceManager struct {
	mu      sync.RWMutex
	records map[string]NamespaceRecord
	subs    map[string]NamespaceSubscriber
}

func NewNamespaceManager() NamespaceManager {
	return &inMemoryNamespaceManager{
		records: make(map[string]NamespaceRecord),
		subs:    make(map[string]NamespaceSubscriber),
	}
}

func NewBootstrappedNamespaceManager(resolver *NamespaceResolver, source NamespaceSource, observedAt time.Time) NamespaceManager {
	manager := NewNamespaceManager()
	if resolver == nil {
		return manager
	}
	manager.Bootstrap(resolver.Snapshot(), source, observedAt)
	return manager
}

func NewWatcherNamespaceManager(ctx context.Context, namespaces []string, source NamespaceSource, observedAt time.Time, subscribers ...NamespaceSubscriber) (NamespaceManager, error) {
	manager := NewNamespaceManager()
	for _, subscriber := range subscribers {
		if subscriber == nil {
			continue
		}
		manager.Subscribe(subscriber)
	}
	_, err := manager.BootstrapAndDispatch(ctx, namespaces, source, observedAt)
	return manager, err
}

func NewDefaultManagedNamespaceWatcherConfig(component string, subscriber NamespaceSubscriber) ManagedNamespaceWatcherConfig {
	return ManagedNamespaceWatcherConfig{
		Component:       component,
		Namespaces:      DefaultNSResolver().Snapshot(),
		RemovalStrategy: NamespaceRemovalStrategyTrackOnly,
		Subscriber:      subscriber,
	}
}

func namespaceManagerLogger(logger *zap.Logger) *zap.Logger {
	if logger != nil {
		return logger
	}
	return zap.NewNop()
}

func PrepareManagedNamespaceWatcher(ctx context.Context, logger *zap.Logger, config ManagedNamespaceWatcherConfig) (NamespaceManager, k8sCache.ResourceEventHandlerFuncs, error) {
	logger = namespaceManagerLogger(logger)
	strategy := config.RemovalStrategy
	if strategy == "" {
		strategy = NamespaceRemovalStrategyTrackOnly
	}
	manager, err := NewWatcherNamespaceManager(ctx, config.Namespaces, NamespaceSourceEnv, time.Now().UTC(), config.Subscriber)
	handlers := NewNamespaceWatcherEventHandlers(ctx, logger, config.Component, manager, strategy)
	LogNamespaceManagerSummary(logger, config.Component+": prepared namespace manager", manager.Summary())
	return manager, handlers, err
}

func RunManagedNamespaceWatcher(ctx context.Context, logger *zap.Logger, kubeClient kubernetes.Interface, mgr managerPkg.Interface, config ManagedNamespaceWatcherConfig) (NamespaceManager, error) {
	logger = namespaceManagerLogger(logger)
	manager, handlers, err := PrepareManagedNamespaceWatcher(ctx, logger, config)
	StartManagedNamespaceWatcher(ctx, logger, config.Component, kubeClient, mgr, handlers)
	// Start reconciler: retries namespaces stuck in NamespacePhaseFailed every 30s.
	mgr.Add(ctx, func(ctx context.Context) {
		logger.Info(config.Component + ": namespace reconciler started")
		manager.RunReconciler(ctx, logger)
		logger.Info(config.Component + ": namespace reconciler stopped")
	})
	LogNamespaceManagerSummary(logger, config.Component+": started namespace watcher", manager.Summary())
	return manager, err
}

func LogNamespaceManagerSummary(logger *zap.Logger, message string, summary NamespaceManagerSummary) {
	if logger == nil {
		return
	}
	logger.Info(message,
		zap.Int("total_namespaces", summary.TotalNamespaces),
		zap.Int("live_namespaces", summary.LiveNamespaces),
		zap.Bool("has_active_namespaces", summary.HasActiveNamespaces()),
		zap.Any("phase_counts", summary.PhaseCounts),
		zap.Any("source_counts", summary.SourceCounts),
		zap.Strings("subscribers", summary.Subscribers),
	)
}

func NamespaceBecameUnmanaged(oldNamespace *corev1.Namespace, newNamespace *corev1.Namespace) bool {
	if oldNamespace == nil || newNamespace == nil {
		return false
	}
	return IsManagedNamespace(oldNamespace.Labels) && !IsManagedNamespace(newNamespace.Labels)
}

func DispatchNamespaceAdd(ctx context.Context, manager NamespaceManager, namespace *corev1.Namespace, source NamespaceSource, observedAt time.Time) (NamespaceRecord, bool, error) {
	if manager == nil || namespace == nil || namespace.Name == "" {
		return NamespaceRecord{}, false, nil
	}
	manager.Upsert(NamespaceEventFromNamespace(NamespaceEventAdd, namespace, source, observedAt))
	return manager.DispatchAdd(ctx, namespace.Name)
}

func DispatchNamespaceResync(ctx context.Context, manager NamespaceManager, namespace *corev1.Namespace, source NamespaceSource, observedAt time.Time) (NamespaceRecord, bool, error) {
	if manager == nil || namespace == nil || namespace.Name == "" {
		return NamespaceRecord{}, false, nil
	}
	manager.Upsert(NamespaceEventFromNamespace(NamespaceEventUpdate, namespace, source, observedAt))
	return manager.DispatchResync(ctx, namespace.Name)
}

func RecordNamespaceRemoval(manager NamespaceManager, obj interface{}, source NamespaceSource, observedAt time.Time) (NamespaceRecord, bool) {
	if manager == nil {
		return NamespaceRecord{}, false
	}
	event := NamespaceEventFromObject(NamespaceEventRemove, obj, source, observedAt)
	if event.Name == "" {
		return NamespaceRecord{}, false
	}
	record := manager.Upsert(event)
	return record, true
}

func HandleWatcherNamespaceAdd(ctx context.Context, logger *zap.Logger, component string, manager NamespaceManager, namespace *corev1.Namespace) {
	logger = namespaceManagerLogger(logger)
	if namespace == nil || namespace.Name == "" {
		return
	}
	if _, _, err := DispatchNamespaceAdd(ctx, manager, namespace, NamespaceSourceWatcher, time.Now().UTC()); err != nil {
		logger.Error(component+": DispatchAdd failed", zap.String("namespace", namespace.Name), zap.Error(err))
		return
	}
	LogNamespaceManagerSummary(logger, component+": namespace manager summary", manager.Summary())
}

func HandleWatcherNamespaceUpdate(ctx context.Context, logger *zap.Logger, component string, manager NamespaceManager, oldNamespace *corev1.Namespace, newNamespace *corev1.Namespace, strategy NamespaceRemovalStrategy) {
	logger = namespaceManagerLogger(logger)
	if newNamespace == nil {
		return
	}
	if NamespaceBecameUnmanaged(oldNamespace, newNamespace) {
		HandleWatcherNamespaceRemoval(ctx, logger, component, manager, newNamespace, strategy)
		return
	}
	if !IsManagedNamespace(newNamespace.Labels) {
		return
	}
	if _, _, err := DispatchNamespaceResync(ctx, manager, newNamespace, NamespaceSourceWatcher, time.Now().UTC()); err != nil {
		logger.Error(component+": DispatchResync failed", zap.String("namespace", newNamespace.Name), zap.Error(err))
		return
	}
	LogNamespaceManagerSummary(logger, component+": namespace manager summary", manager.Summary())
}

func HandleWatcherNamespaceDelete(ctx context.Context, logger *zap.Logger, component string, manager NamespaceManager, obj interface{}, strategy NamespaceRemovalStrategy) {
	HandleWatcherNamespaceRemoval(ctx, logger, component, manager, obj, strategy)
}

func HandleWatcherNamespaceRemoval(ctx context.Context, logger *zap.Logger, component string, manager NamespaceManager, obj interface{}, strategy NamespaceRemovalStrategy) {
	logger = namespaceManagerLogger(logger)
	record, ok := RecordNamespaceRemoval(manager, obj, NamespaceSourceWatcher, time.Now().UTC())
	if !ok {
		return
	}
	// Always remove from global resolver so Snapshot() and idleObjectReaper
	// stop iterating this namespace. This is safe: if the same name is re-added
	// later, AddNamespace will return true and all subscribers will re-register.
	DefaultNSResolver().RemoveNamespace(record.Name)

	if strategy == NamespaceRemovalStrategyDispatchRemove {
		if _, _, err := manager.DispatchRemove(ctx, record.Name); err != nil {
			logger.Error(component+": DispatchRemove failed", zap.String("namespace", record.Name), zap.Error(err))
			return
		}
		logger.Info(component+": namespace removal dispatched through subscribers",
			zap.String("namespace", record.Name))
		LogNamespaceManagerSummary(logger, component+": namespace manager summary", manager.Summary())
		return
	}
	logger.Info(component+": namespace removed from manager state; runtime registrations kept",
		zap.String("namespace", record.Name))
	LogNamespaceManagerSummary(logger, component+": namespace manager summary", manager.Summary())
}

func NewNamespaceWatcherEventHandlers(ctx context.Context, logger *zap.Logger, component string, manager NamespaceManager, strategy NamespaceRemovalStrategy) k8sCache.ResourceEventHandlerFuncs {
	logger = namespaceManagerLogger(logger)
	return k8sCache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			namespace, ok := obj.(*corev1.Namespace)
			if ok {
				HandleWatcherNamespaceAdd(ctx, logger, component, manager, namespace)
			}
		},
		UpdateFunc: func(oldObj, newObj interface{}) {
			oldNamespace, _ := oldObj.(*corev1.Namespace)
			newNamespace, ok := newObj.(*corev1.Namespace)
			if ok {
				HandleWatcherNamespaceUpdate(ctx, logger, component, manager, oldNamespace, newNamespace, strategy)
			}
		},
		DeleteFunc: func(obj interface{}) {
			HandleWatcherNamespaceDelete(ctx, logger, component, manager, obj, strategy)
		},
	}
}

func StartManagedNamespaceWatcher(ctx context.Context, logger *zap.Logger, component string, kubeClient kubernetes.Interface, mgr managerPkg.Interface, handlers k8sCache.ResourceEventHandlerFuncs) {
	logger = namespaceManagerLogger(logger)
	factory := k8sInformers.NewSharedInformerFactoryWithOptions(
		kubeClient,
		30*time.Minute,
		k8sInformers.WithTweakListOptions(func(opts *metav1.ListOptions) {
			opts.LabelSelector = ManagedNamespaceLabelSelector()
		}),
	)

	nsInformer := factory.Core().V1().Namespaces().Informer()
	_, _ = nsInformer.AddEventHandler(handlers)

	mgr.Add(ctx, func(ctx context.Context) {
		logger.Info(component+": started", zap.String("label", ManagedNamespaceLabelSelector()))
		factory.Start(ctx.Done())
		factory.WaitForCacheSync(ctx.Done())
		logger.Info(component + ": cache synced — watching for new namespaces")
		<-ctx.Done()
		logger.Info(component + ": stopped")
	})
}

func (m *inMemoryNamespaceManager) Summary() NamespaceManagerSummary {
	records := m.SnapshotRecords()
	summary := NamespaceManagerSummary{
		TotalNamespaces: len(records),
		LiveNamespaces:  len(m.Snapshot()),
		PhaseCounts:     make(map[NamespacePhase]int),
		SourceCounts:    make(map[NamespaceSource]int),
		Subscribers:     m.SnapshotSubscribers(),
	}
	if summary.Subscribers == nil {
		summary.Subscribers = []string{}
	}
	for _, record := range records {
		summary.PhaseCounts[record.Phase]++
		summary.SourceCounts[record.Source]++
	}
	return summary
}

func (m *inMemoryNamespaceManager) Subscribe(subscriber NamespaceSubscriber) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.subs[subscriber.Name()] = subscriber
}

func (m *inMemoryNamespaceManager) Bootstrap(namespaces []string, source NamespaceSource, observedAt time.Time) []NamespaceRecord {
	records := make([]NamespaceRecord, 0, len(namespaces))
	for _, namespace := range namespaces {
		records = append(records, m.Upsert(NamespaceEvent{
			Type:       NamespaceEventAdd,
			Name:       namespace,
			Source:     source,
			ObservedAt: observedAt,
		}))
	}
	sort.Slice(records, func(i, j int) bool {
		return records[i].Name < records[j].Name
	})
	return records
}

func (m *inMemoryNamespaceManager) BootstrapAndDispatch(ctx context.Context, namespaces []string, source NamespaceSource, observedAt time.Time) ([]NamespaceRecord, error) {
	records := m.Bootstrap(namespaces, source, observedAt)
	var joinErr error
	for _, record := range records {
		updatedRecord, ok, err := m.DispatchAdd(ctx, record.Name)
		if ok {
			record = updatedRecord
		}
		if err != nil {
			joinErr = errors.Join(joinErr, err)
		}
		for index := range records {
			if records[index].Name == record.Name {
				records[index] = record
				break
			}
		}
	}
	return records, joinErr
}

func (m *inMemoryNamespaceManager) SnapshotSubscribers() []string {
	m.mu.RLock()
	defer m.mu.RUnlock()

	names := make([]string, 0, len(m.subs))
	for name := range m.subs {
		names = append(names, name)
	}
	sort.Strings(names)
	return names
}

func (m *inMemoryNamespaceManager) snapshotSubscriberObjects() []NamespaceSubscriber {
	m.mu.RLock()
	defer m.mu.RUnlock()

	names := make([]string, 0, len(m.subs))
	for name := range m.subs {
		names = append(names, name)
	}
	sort.Strings(names)

	subscribers := make([]NamespaceSubscriber, 0, len(names))
	for _, name := range names {
		subscribers = append(subscribers, m.subs[name])
	}
	return subscribers
}

func (m *inMemoryNamespaceManager) Snapshot() []string {
	m.mu.RLock()
	defer m.mu.RUnlock()

	namespaces := make([]string, 0, len(m.records))
	for name, record := range m.records {
		if record.Phase == NamespacePhaseRemoved {
			continue
		}
		namespaces = append(namespaces, name)
	}
	sort.Strings(namespaces)
	return namespaces
}

func (m *inMemoryNamespaceManager) SnapshotRecords() []NamespaceRecord {
	m.mu.RLock()
	defer m.mu.RUnlock()

	records := make([]NamespaceRecord, 0, len(m.records))
	for _, record := range m.records {
		records = append(records, record.Clone())
	}
	sort.Slice(records, func(i, j int) bool {
		return records[i].Name < records[j].Name
	})
	return records
}

func (m *inMemoryNamespaceManager) Get(name string) (NamespaceRecord, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	record, ok := m.records[name]
	if !ok {
		return NamespaceRecord{}, false
	}
	return record.Clone(), true
}

func (m *inMemoryNamespaceManager) Upsert(event NamespaceEvent) NamespaceRecord {
	m.mu.Lock()
	defer m.mu.Unlock()

	now := event.ObservedAt
	if now.IsZero() {
		now = time.Now().UTC()
	}

	record, exists := m.records[event.Name]
	if !exists {
		record = NamespaceRecord{
			Name:            event.Name,
			RegisteredParts: make(map[string]NamespacePartState),
		}
		if event.Type == NamespaceEventRemove {
			record.Generation = 0
		} else {
			record.Generation = 1
		}
	} else if event.Type != NamespaceEventRemove {
		record.Generation++
	}

	record.Name = event.Name
	record.Source = event.Source
	record.Labels = cloneStringMap(event.Labels)
	record.UpdatedAt = now

	switch event.Type {
	case NamespaceEventAdd, NamespaceEventUpdate, NamespaceEventResync:
		record.Phase = NamespacePhaseDiscovered
		record.LastError = ""
	case NamespaceEventRemove:
		record.Phase = NamespacePhaseRemoved
	}

	if record.RegisteredParts == nil {
		record.RegisteredParts = make(map[string]NamespacePartState)
	}

	m.records[event.Name] = record
	return record.Clone()
}

func (m *inMemoryNamespaceManager) DispatchAdd(ctx context.Context, namespace string) (NamespaceRecord, bool, error) {
	return m.dispatch(ctx, namespace, func(subscriber NamespaceSubscriber, record NamespaceRecord) error {
		return subscriber.OnNamespaceAdd(ctx, record)
	})
}

func (m *inMemoryNamespaceManager) DispatchRemove(ctx context.Context, namespace string) (NamespaceRecord, bool, error) {
	record, ok := m.Get(namespace)
	if !ok {
		return NamespaceRecord{}, false, nil
	}

	var firstErr error
	for _, subscriber := range m.snapshotSubscriberObjects() {
		err := subscriber.OnNamespaceRemove(ctx, record)
		if err != nil && firstErr == nil {
			firstErr = err
		}
	}

	removedRecord := m.Upsert(NamespaceEvent{
		Type:       NamespaceEventRemove,
		Name:       record.Name,
		Labels:     record.Labels,
		Source:     record.Source,
		ObservedAt: time.Now().UTC(),
	})
	return removedRecord, true, firstErr
}

func (m *inMemoryNamespaceManager) DispatchResync(ctx context.Context, namespace string) (NamespaceRecord, bool, error) {
	return m.dispatch(ctx, namespace, func(subscriber NamespaceSubscriber, record NamespaceRecord) error {
		return subscriber.OnNamespaceResync(ctx, record)
	})
}

func (m *inMemoryNamespaceManager) dispatch(ctx context.Context, namespace string, handler func(NamespaceSubscriber, NamespaceRecord) error) (NamespaceRecord, bool, error) {
	_, ok := m.Get(namespace)
	if !ok {
		return NamespaceRecord{}, false, nil
	}

	subscribers := m.snapshotSubscriberObjects()
	for _, sub := range subscribers {
		_, _ = m.MarkPartRegistering(namespace, sub.Name())
	}

	// Run all subscriber handlers in parallel — each handler makes independent k8s API calls.
	// MarkPart* methods are internally mutex-protected and safe for concurrent calls.
	var (
		wg   sync.WaitGroup
		mu   sync.Mutex
		errs error
	)
	for _, sub := range subscribers {
		sub := sub
		wg.Add(1)
		go func() {
			defer wg.Done()
			currentRecord, _ := m.Get(namespace)
			err := handler(sub, currentRecord)
			if err != nil {
				_, _ = m.MarkPartFailed(namespace, sub.Name(), err)
				mu.Lock()
				errs = errors.Join(errs, err)
				mu.Unlock()
				return
			}
			_, _ = m.MarkPartActive(namespace, sub.Name())
		}()
	}
	wg.Wait()

	record, _ := m.Get(namespace)
	return record, true, errs
}

func (m *inMemoryNamespaceManager) MarkPartState(namespace string, part string, state NamespacePartState) (NamespaceRecord, bool) {
	m.mu.Lock()
	defer m.mu.Unlock()

	record, ok := m.records[namespace]
	if !ok {
		return NamespaceRecord{}, false
	}
	if record.RegisteredParts == nil {
		record.RegisteredParts = make(map[string]NamespacePartState)
	}
	if state.UpdatedAt.IsZero() {
		state.UpdatedAt = time.Now().UTC()
	}
	record.RegisteredParts[part] = state
	record.Phase = deriveNamespacePhase(record)
	record.UpdatedAt = state.UpdatedAt
	m.records[namespace] = record
	return record.Clone(), true
}

func (m *inMemoryNamespaceManager) MarkPartRegistering(namespace string, part string) (NamespaceRecord, bool) {
	return m.MarkPartState(namespace, part, NamespacePartState{State: NamespacePartStateRegistering})
}

func (m *inMemoryNamespaceManager) MarkPartActive(namespace string, part string) (NamespaceRecord, bool) {
	return m.MarkPartState(namespace, part, NamespacePartState{State: NamespacePartStateActive})
}

func (m *inMemoryNamespaceManager) MarkPartFailed(namespace string, part string, err error) (NamespaceRecord, bool) {
	lastError := ""
	if err != nil {
		lastError = err.Error()
	}
	return m.MarkPartState(namespace, part, NamespacePartState{State: NamespacePartStateFailed, LastError: lastError})
}

func (m *inMemoryNamespaceManager) Remove(name string) bool {
	m.mu.Lock()
	defer m.mu.Unlock()

	if _, ok := m.records[name]; !ok {
		return false
	}
	delete(m.records, name)
	return true
}

// RunReconciler periodically finds namespaces in NamespacePhaseFailed and retries them
// via DispatchResync. This ensures transient k8s API errors (e.g. temporary 503 on
// EnsureNamespaceSA or executor-type AddNamespace) do not permanently strand a namespace.
// logger receives one log line per retry attempt and per outcome.
// Exits when ctx is cancelled.
func (m *inMemoryNamespaceManager) RunReconciler(ctx context.Context, logger *zap.Logger) {
	if logger == nil {
		logger = zap.NewNop()
	}
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			m.mu.RLock()
			var failedNS []string
			for ns, rec := range m.records {
				if rec.Phase == NamespacePhaseFailed {
					failedNS = append(failedNS, ns)
				}
			}
			m.mu.RUnlock()
			if len(failedNS) == 0 {
				continue
			}
			logger.Info("namespace reconciler: retrying failed namespaces",
				zap.Int("count", len(failedNS)),
				zap.Strings("namespaces", failedNS),
			)
			for _, ns := range failedNS {
				logger.Info("namespace reconciler: dispatching resync", zap.String("namespace", ns))
				_, _, err := m.DispatchResync(ctx, ns)
				if err != nil {
					logger.Error("namespace reconciler: resync still failing, will retry",
						zap.String("namespace", ns),
						zap.Error(err),
					)
				} else {
					logger.Info("namespace reconciler: resync succeeded", zap.String("namespace", ns))
				}
			}
		}
	}
}

func deriveNamespacePhase(record NamespaceRecord) NamespacePhase {
	if len(record.RegisteredParts) == 0 {
		return record.Phase
	}

	hasRegistering := false
	for _, part := range record.RegisteredParts {
		switch part.State {
		case NamespacePartStateFailed:
			return NamespacePhaseFailed
		case NamespacePartStateRegistering:
			hasRegistering = true
		case NamespacePartStateActive:
			continue
		default:
			hasRegistering = true
		}
	}

	if hasRegistering {
		return NamespacePhaseRegistering
	}

	return NamespacePhaseActive
}