feat(unified-storage): add ring integration for sub-index sharding

Phase 2 of unified storage search sharding implementation:

- Add OwnsSubIndex() to service.go that includes subIndexID in ring hash
- Modify ring hash: fmt.Sprintf("%s/%d", namespace, subIndexID)
- Update OwnsIndex() to delegate to OwnsSubIndex(key, 0) for compatibility
- Add OwnsSubIndex callback to BleveOptions for ownership checks
- Update eviction logic to handle sub-indexes separately:
  - runEvictExpiredOrUnownedIndexes() now processes subIndexCache
  - Uses ownsSubIndexFn to check sub-index ownership via ring
- Add closeSubIndex() helper for proper sub-index cleanup
- Update closeAllIndexes() to close both main and sub-indexes
- Pass SubIndexCount and LargeFolderThreshold from config to BleveOptions

This enables sub-indexes to be distributed across ring nodes, with each
sub-index potentially owned by a different node based on the ring hash.

pkg/storage/unified/search/bleve.go

@@ -83,6 +83,11 @@ type BleveOptions struct {
 	// If nil, all indexes are owned by the current instance.
 	OwnsIndex func(key resource.NamespacedResource) (bool, error)
 
+	// OwnsSubIndex is called to check whether a specific sub-index is owned by the current instance.
+	// This function considers the sub-index ID when determining ownership via ring hash.
+	// If nil, falls back to OwnsIndex behavior with subIndexID=0.
+	OwnsSubIndex func(key resource.NamespacedResource, subIndexID int) (bool, error)
+
 	// SubIndexCount is the number of sub-indexes per (namespace, group, resource).
 	// When > 0, documents are distributed across sub-indexes using consistent hashing.
 	// This enables horizontal scaling for large namespaces (1M+ documents).
@@ -102,6 +107,10 @@ type bleveBackend struct {
 	// set from opts.OwnsIndex, always non-nil
 	ownsIndexFn func(key resource.NamespacedResource) (bool, error)
 
+	// set from opts.OwnsSubIndex, always non-nil
+	// Used for checking ownership of sub-indexes when sharding is enabled
+	ownsSubIndexFn func(key resource.NamespacedResource, subIndexID int) (bool, error)
+
 	cacheMx sync.RWMutex
 	cache   map[resource.NamespacedResource]*bleveIndex
 
@@ -152,13 +161,23 @@ func NewBleveBackend(opts BleveOptions, indexMetrics *resource.BleveIndexMetrics
 		ownFn = func(key resource.NamespacedResource) (bool, error) { return true, nil }
 	}
 
+	ownSubFn := opts.OwnsSubIndex
+	if ownSubFn == nil {
+		// By default, fall back to OwnsIndex behavior (ignore subIndexID).
+		// This maintains backward compatibility when sub-index sharding is not enabled.
+		ownSubFn = func(key resource.NamespacedResource, subIndexID int) (bool, error) {
+			return ownFn(key)
+		}
+	}
+
 	be := &bleveBackend{
-		log:           l,
-		cache:         map[resource.NamespacedResource]*bleveIndex{},
-		subIndexCache: map[resource.SubIndexKey]*bleveIndex{},
-		opts:          opts,
-		ownsIndexFn:   ownFn,
-		indexMetrics:  indexMetrics,
+		log:            l,
+		cache:          map[resource.NamespacedResource]*bleveIndex{},
+		subIndexCache:  map[resource.SubIndexKey]*bleveIndex{},
+		opts:           opts,
+		ownsIndexFn:    ownFn,
+		ownsSubIndexFn: ownSubFn,
+		indexMetrics:   indexMetrics,
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -292,7 +311,13 @@ func (b *bleveBackend) runEvictExpiredOrUnownedIndexes(now time.Time) {
 	unowned := map[resource.NamespacedResource]*bleveIndex{}
 	ownCheckErrors := map[resource.NamespacedResource]error{}
 
+	// For sub-indexes
+	expiredSubIndexes := map[resource.SubIndexKey]*bleveIndex{}
+	unownedSubIndexes := map[resource.SubIndexKey]*bleveIndex{}
+	ownSubCheckErrors := map[resource.SubIndexKey]error{}
+
 	b.cacheMx.Lock()
+	// Process main cache (non-sharded indexes)
 	for key, idx := range b.cache {
 		// Check if index has expired.
 		if !idx.expiration.IsZero() && now.After(idx.expiration) {
@@ -312,21 +337,75 @@ func (b *bleveBackend) runEvictExpiredOrUnownedIndexes(now time.Time) {
 			}
 		}
 	}
+
+	// Process sub-index cache (sharded indexes)
+	for subKey, idx := range b.subIndexCache {
+		// Check if sub-index has expired.
+		if !idx.expiration.IsZero() && now.After(idx.expiration) {
+			delete(b.subIndexCache, subKey)
+			expiredSubIndexes[subKey] = idx
+			continue
+		}
+
+		// Check if sub-index is owned by this instance using OwnsSubIndex.
+		// This considers the subIndexID when determining ownership via ring hash.
+		if cacheTTLMillis > 0 {
+			owned, err := b.ownsSubIndexFn(subKey.NamespacedResource, subKey.SubIndexID)
+			if err != nil {
+				ownSubCheckErrors[subKey] = err
+			} else if !owned && now.UnixMilli()-idx.lastFetchedFromCache.Load() > cacheTTLMillis {
+				delete(b.subIndexCache, subKey)
+				unownedSubIndexes[subKey] = idx
+			}
+		}
+	}
 	b.cacheMx.Unlock()
 
+	// Log errors for main cache ownership checks
 	for key, err := range ownCheckErrors {
 		b.log.Warn("failed to check if index belongs to this instance", "key", key, "err", err)
 	}
 
+	// Log errors for sub-index ownership checks
+	for subKey, err := range ownSubCheckErrors {
+		b.log.Warn("failed to check if sub-index belongs to this instance", "subKey", subKey, "err", err)
+	}
+
+	// Evict unowned main indexes
 	for key, idx := range unowned {
 		b.log.Info("index evicted from cache", "reason", "unowned", "key", key, "storage", idx.indexStorage)
 		b.closeIndex(idx, key)
 	}
 
+	// Evict unowned sub-indexes
+	for subKey, idx := range unownedSubIndexes {
+		b.log.Info("sub-index evicted from cache", "reason", "unowned", "subKey", subKey, "storage", idx.indexStorage)
+		b.closeSubIndex(idx, subKey)
+	}
+
+	// Evict expired main indexes
 	for key, idx := range expired {
 		b.log.Info("index evicted from cache", "reason", "expired", "key", key, "storage", idx.indexStorage)
 		b.closeIndex(idx, key)
 	}
+
+	// Evict expired sub-indexes
+	for subKey, idx := range expiredSubIndexes {
+		b.log.Info("sub-index evicted from cache", "reason", "expired", "subKey", subKey, "storage", idx.indexStorage)
+		b.closeSubIndex(idx, subKey)
+	}
+}
+
+// closeSubIndex closes a sub-index and updates metrics.
+func (b *bleveBackend) closeSubIndex(idx *bleveIndex, key resource.SubIndexKey) {
+	err := idx.stopUpdaterAndCloseIndex()
+	if err != nil {
+		b.log.Error("failed to close sub-index", "key", key, "err", err)
+	}
+
+	if b.indexMetrics != nil {
+		b.indexMetrics.OpenIndexes.WithLabelValues(idx.indexStorage).Dec()
+	}
 }
 
 // updateIndexSizeMetric sets the total size of all file-based indices metric.
@@ -742,6 +821,7 @@ func (b *bleveBackend) closeAllIndexes() {
 	b.cacheMx.Lock()
 	defer b.cacheMx.Unlock()
 
+	// Close main indexes
 	for key, idx := range b.cache {
 		if err := idx.stopUpdaterAndCloseIndex(); err != nil {
 			b.log.Error("Failed to close index", "err", err)
@@ -752,6 +832,18 @@ func (b *bleveBackend) closeAllIndexes() {
 			b.indexMetrics.OpenIndexes.WithLabelValues(idx.indexStorage).Dec()
 		}
 	}
+
+	// Close sub-indexes
+	for subKey, idx := range b.subIndexCache {
+		if err := idx.stopUpdaterAndCloseIndex(); err != nil {
+			b.log.Error("Failed to close sub-index", "subKey", subKey, "err", err)
+		}
+		delete(b.subIndexCache, subKey)
+
+		if b.indexMetrics != nil {
+			b.indexMetrics.OpenIndexes.WithLabelValues(idx.indexStorage).Dec()
+		}
+	}
 }
 
 type updateRequest struct {

pkg/storage/unified/search/options.go

@@ -17,6 +17,7 @@ func NewSearchOptions(
 	docs resource.DocumentBuilderSupplier,
 	indexMetrics *resource.BleveIndexMetrics,
 	ownsIndexFn func(key resource.NamespacedResource) (bool, error),
+	ownsSubIndexFn ...func(key resource.NamespacedResource, subIndexID int) (bool, error),
 ) (resource.SearchOptions, error) {
 	//nolint:staticcheck // not yet migrated to OpenFeature
 	if cfg.EnableSearch || features.IsEnabledGlobally(featuremgmt.FlagProvisioning) {
@@ -39,13 +40,22 @@ func NewSearchOptions(
 			}
 		}
 
+		// Get OwnsSubIndex function if provided
+		var ownsSubIdx func(key resource.NamespacedResource, subIndexID int) (bool, error)
+		if len(ownsSubIndexFn) > 0 && ownsSubIndexFn[0] != nil {
+			ownsSubIdx = ownsSubIndexFn[0]
+		}
+
 		bleve, err := NewBleveBackend(BleveOptions{
 			Root:                   root,
 			FileThreshold:          int64(cfg.IndexFileThreshold), // fewer than X items will use a memory index
 			IndexCacheTTL:          cfg.IndexCacheTTL,             // How long to keep the index cache in memory
 			BuildVersion:           cfg.BuildVersion,
 			OwnsIndex:              ownsIndexFn,
+			OwnsSubIndex:           ownsSubIdx,
 			IndexMinUpdateInterval: cfg.IndexMinUpdateInterval,
+			SubIndexCount:          cfg.SubIndexesPerNamespace,
+			LargeFolderThreshold:   cfg.LargeFolderThreshold,
 		}, indexMetrics)
 
 		if err != nil {

pkg/storage/unified/sql/service.go

@@ -226,6 +226,18 @@ var (
 )
 
 func (s *service) OwnsIndex(key resource.NamespacedResource) (bool, error) {
+	// When sub-index sharding is enabled, use OwnsSubIndex with subIndexID=0
+	// to maintain backward compatibility. OwnsIndex is used for the main index.
+	return s.OwnsSubIndex(key, 0)
+}
+
+// OwnsSubIndex checks if the current instance owns a specific sub-index.
+// The sub-index ID is included in the ring hash to distribute sub-indexes
+// across nodes in the ring. This enables horizontal scaling for large namespaces.
+//
+// When subIndexID is 0 and SubIndexesPerNamespace is 0 (disabled), this behaves
+// exactly like the original OwnsIndex - maintaining backward compatibility.
+func (s *service) OwnsSubIndex(key resource.NamespacedResource, subIndexID int) (bool, error) {
 	if s.searchRing == nil {
 		return true, nil
 	}
@@ -235,9 +247,22 @@ func (s *service) OwnsIndex(key resource.NamespacedResource) (bool, error) {
 	}
 
 	ringHasher := fnv.New32a()
-	_, err := ringHasher.Write([]byte(key.Namespace))
-	if err != nil {
-		return false, fmt.Errorf("error hashing namespace: %w", err)
+
+	// When sub-index sharding is enabled (SubIndexesPerNamespace > 0),
+	// include the subIndexID in the hash to distribute sub-indexes across nodes.
+	// This allows different sub-indexes of the same namespace to be owned by
+	// different nodes, enabling horizontal scaling.
+	if s.cfg.SubIndexesPerNamespace > 0 {
+		_, err := ringHasher.Write([]byte(fmt.Sprintf("%s/%d", key.Namespace, subIndexID)))
+		if err != nil {
+			return false, fmt.Errorf("error hashing namespace with sub-index: %w", err)
+		}
+	} else {
+		// Original behavior: hash only the namespace
+		_, err := ringHasher.Write([]byte(key.Namespace))
+		if err != nil {
+			return false, fmt.Errorf("error hashing namespace: %w", err)
+		}
 	}
 
 	rs, err := s.searchRing.GetWithOptions(ringHasher.Sum32(), searchOwnerRead, ring.WithReplicationFactor(s.searchRing.ReplicationFactor()))
@@ -261,7 +286,7 @@ func (s *service) starting(ctx context.Context) error {
 		return err
 	}
 
-	searchOptions, err := search.NewSearchOptions(s.features, s.cfg, s.docBuilders, s.indexMetrics, s.OwnsIndex)
+	searchOptions, err := search.NewSearchOptions(s.features, s.cfg, s.docBuilders, s.indexMetrics, s.OwnsIndex, s.OwnsSubIndex)
 	if err != nil {
 		return err
 	}