opensandbox-group--opensandbox/kubernetes/internal/controller/allocator.go

// Copyright 2025 Alibaba Group Holding Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package controller

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"sync"

	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	logf "sigs.k8s.io/controller-runtime/pkg/log"

	sandboxv1alpha1 "github.com/alibaba/OpenSandbox/sandbox-k8s/apis/sandbox/v1alpha1"
	algorithm "github.com/alibaba/OpenSandbox/sandbox-k8s/internal/controller/algorithm"
	"github.com/alibaba/OpenSandbox/sandbox-k8s/internal/utils"
)

type AllocationStore interface {
	GetAllocation(ctx context.Context, pool *sandboxv1alpha1.Pool) (*PoolAllocation, error)
	SetAllocation(ctx context.Context, pool *sandboxv1alpha1.Pool, allocation *PoolAllocation) error
	ClearAllocation(ctx context.Context, ns string, poolName string) error
	ReleaseAllocation(ctx context.Context, ns string, poolName string, pods []string)
	UpdateAllocation(ctx context.Context, ns string, poolName string, sandboxName string, pods []string)
	// ReleaseSandboxAllocation releases all pods allocated to the given sandbox name.
	// This is used when a BatchSandbox is deleted and its allocation needs to be cleaned up.
	ReleaseSandboxAllocation(ctx context.Context, ns string, poolName string, sandboxName string)
	Recover(ctx context.Context, c client.Client) error
}

// poolEntry represents a single pool's allocation data with its own lock for fine-grained concurrency control
type poolEntry struct {
	mu   sync.RWMutex
	data map[string]string // podName -> sandboxName
}

// InMemoryAllocationStore depends on annoAllocationSyncer to get allocation info from BatchSandbox.
type InMemoryAllocationStore struct {
	poolsMu sync.RWMutex
	pools   map[string]*poolEntry
	syncer  *annoAllocationSyncer
}

func NewInMemoryAllocationStore() AllocationStore {
	return &InMemoryAllocationStore{
		pools:  make(map[string]*poolEntry),
		syncer: &annoAllocationSyncer{},
	}
}

// Recover builds the allocation map from all BatchSandboxes
// This should be called once during controller initialization before reconcile starts
func (store *InMemoryAllocationStore) Recover(ctx context.Context, c client.Client) error {
	log := logf.FromContext(ctx)
	log.Info("Starting allocation recovery from BatchSandboxes")

	batchSandboxList := &sandboxv1alpha1.BatchSandboxList{}
	if err := c.List(ctx, batchSandboxList); err != nil {
		return fmt.Errorf("failed to list batch sandboxes for recovery: %w", err)
	}

	// Build new pools map first without holding the lock
	newPools := make(map[string]*poolEntry)

	for _, sbx := range batchSandboxList.Items {
		poolRef := sbx.Spec.PoolRef
		if poolRef == "" {
			continue
		}
		allocation, err := store.syncer.GetAllocation(ctx, &sbx)
		if err != nil {
			log.Error(err, "Failed to unmarshal sandbox allocation during recovery", "sandbox", sbx.Name)
			return err
		}
		key := store.poolKey(sbx.Namespace, poolRef)
		entry, exists := newPools[key]
		if !exists {
			entry = &poolEntry{
				data: make(map[string]string),
			}
			newPools[key] = entry
		}

		for _, podName := range allocation.Pods {
			entry.data[podName] = sbx.Name
		}
		// Filter pods that have already been released (alloc-released records completed recycle).
		// alloc-release (in-progress) pods must NOT be filtered: the recycle handler is still
		// processing them and they are still "in use" from the pool's perspective.
		allocReleased, err := store.syncer.GetReleased(ctx, &sbx)
		if err != nil {
			log.Error(err, "Failed to unmarshal sandbox released during recovery", "sandbox", sbx.Name)
			return err
		}
		for _, podName := range allocReleased.Pods {
			if entry.data[podName] == sbx.Name {
				delete(entry.data, podName)
			}
		}

		log.Info("Recovered sandbox allocation", "pool", poolRef, "sandbox", sbx.Name, "pods", len(allocation.Pods))
	}

	store.poolsMu.Lock()
	store.pools = newPools
	store.poolsMu.Unlock()

	log.Info("Allocation recovery completed", "totalPools", len(store.pools))
	return nil
}

func (store *InMemoryAllocationStore) ClearAllocation(ctx context.Context, ns string, poolName string) error {
	log := logf.FromContext(ctx)
	store.poolsMu.Lock()
	log.Info("Clearing pool allocation", "namespace", ns, "pool", poolName)
	delete(store.pools, store.poolKey(ns, poolName))
	store.poolsMu.Unlock()
	return nil
}

func (store *InMemoryAllocationStore) GetAllocation(ctx context.Context, pool *sandboxv1alpha1.Pool) (*PoolAllocation, error) {
	store.poolsMu.RLock()
	entry, exists := store.pools[store.poolKey(pool.Namespace, pool.Name)]
	store.poolsMu.RUnlock()

	alloc := &PoolAllocation{
		PodAllocation: make(map[string]string),
	}

	if !exists {
		return alloc, nil
	}

	entry.mu.RLock()
	defer entry.mu.RUnlock()

	for podName, sandboxName := range entry.data {
		alloc.PodAllocation[podName] = sandboxName
	}

	return alloc, nil
}

func (store *InMemoryAllocationStore) SetAllocation(ctx context.Context, pool *sandboxv1alpha1.Pool, alloc *PoolAllocation) error {
	entry := store.getOrCreatePool(pool.Namespace, pool.Name)

	entry.mu.Lock()
	defer entry.mu.Unlock()

	entry.data = make(map[string]string)
	for podName, sandboxName := range alloc.PodAllocation {
		entry.data[podName] = sandboxName
	}

	return nil
}

func (store *InMemoryAllocationStore) ReleaseAllocation(ctx context.Context, ns string, poolName string, pods []string) {
	entry := store.getOrCreatePool(ns, poolName)

	entry.mu.Lock()
	defer entry.mu.Unlock()

	for _, podName := range pods {
		delete(entry.data, podName)
	}
}

func (store *InMemoryAllocationStore) UpdateAllocation(ctx context.Context, ns string, poolName string, sandboxName string, pods []string) {
	entry := store.getOrCreatePool(ns, poolName)

	entry.mu.Lock()
	defer entry.mu.Unlock()

	for podName, sbxName := range entry.data {
		if sbxName == sandboxName {
			delete(entry.data, podName)
		}
	}

	for _, podName := range pods {
		entry.data[podName] = sandboxName
	}
}

// ReleaseSandboxAllocation releases all pods allocated to the given sandbox from the in-memory store.
// This should be called when a BatchSandbox is deleted to ensure the allocation state is cleaned up.
func (store *InMemoryAllocationStore) ReleaseSandboxAllocation(ctx context.Context, ns string, poolName string, sandboxName string) {
	store.poolsMu.RLock()
	entry, exists := store.pools[store.poolKey(ns, poolName)]
	store.poolsMu.RUnlock()

	if !exists {
		return
	}

	entry.mu.Lock()
	defer entry.mu.Unlock()

	for podName, sbxName := range entry.data {
		if sbxName == sandboxName {
			delete(entry.data, podName)
		}
	}
}

// getOrCreatePool returns the pool entry for the given pool name, creating it if necessary.
// This method uses a double-checked locking pattern to ensure thread-safe creation.
func (store *InMemoryAllocationStore) getOrCreatePool(ns string, poolName string) *poolEntry {
	store.poolsMu.RLock()
	entry, exists := store.pools[store.poolKey(ns, poolName)]
	store.poolsMu.RUnlock()

	if exists {
		return entry
	}

	store.poolsMu.Lock()
	defer store.poolsMu.Unlock()

	// Double-check after acquiring write lock
	if entry, exists := store.pools[store.poolKey(ns, poolName)]; exists {
		return entry
	}

	entry = &poolEntry{
		data: make(map[string]string),
	}
	store.pools[store.poolKey(ns, poolName)] = entry
	return entry
}

func (store *InMemoryAllocationStore) poolKey(ns, name string) string {
	return ns + "/" + name
}

type AllocationSyncer interface {
	SetAllocation(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox, allocation *SandboxAllocation) error
	GetAllocation(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox) (*SandboxAllocation, error)
	GetRelease(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox) (*AllocationRelease, error)
	SetReleased(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox, released *AllocationReleased) error
	GetReleased(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox) (*AllocationReleased, error)
}

type annoAllocationSyncer struct {
	client client.Client
}

func NewAnnoAllocationSyncer(client client.Client) AllocationSyncer {
	return &annoAllocationSyncer{
		client: client,
	}
}

func (syncer *annoAllocationSyncer) SetAllocation(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox, allocation *SandboxAllocation) error {
	js, err := json.Marshal(allocation)
	if err != nil {
		return err
	}
	anno := sandbox.GetAnnotations()
	if anno == nil {
		anno = make(map[string]string)
	}
	anno[AnnoAllocStatusKey] = string(js)
	sandbox.SetAnnotations(anno)

	needAddFinalizer := !controllerutil.ContainsFinalizer(sandbox, FinalizerPoolAllocation)
	if needAddFinalizer {
		sandbox.SetFinalizers(append(sandbox.GetFinalizers(), FinalizerPoolAllocation))
	}

	meta := map[string]any{
		"annotations": map[string]string{
			AnnoAllocStatusKey: string(js),
		},
	}
	if needAddFinalizer {
		meta["finalizers"] = sandbox.GetFinalizers()
	}
	patchData, err := json.Marshal(map[string]any{"metadata": meta})
	if err != nil {
		return err
	}
	obj := &sandboxv1alpha1.BatchSandbox{}
	obj.Name = sandbox.Name
	obj.Namespace = sandbox.Namespace
	return syncer.client.Patch(ctx, obj, client.RawPatch(types.MergePatchType, patchData))
}

func (syncer *annoAllocationSyncer) GetAllocation(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox) (*SandboxAllocation, error) {
	allocation := &SandboxAllocation{
		Pods: make([]string, 0),
	}
	anno := sandbox.GetAnnotations()
	if anno == nil {
		return allocation, nil
	}
	if raw := anno[AnnoAllocStatusKey]; raw != "" {
		err := json.Unmarshal([]byte(raw), allocation)
		if err != nil {
			return nil, err
		}
	}
	return allocation, nil
}

func (syncer *annoAllocationSyncer) GetRelease(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox) (*AllocationRelease, error) {
	release := &AllocationRelease{
		Pods: make([]string, 0),
	}
	anno := sandbox.GetAnnotations()
	if anno == nil {
		return release, nil
	}
	if raw := anno[AnnoAllocReleaseKey]; raw != "" {
		err := json.Unmarshal([]byte(raw), release)
		if err != nil {
			return nil, err
		}
	}
	return release, nil
}

func (syncer *annoAllocationSyncer) GetReleased(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox) (*AllocationReleased, error) {
	released := &AllocationReleased{
		Pods: make([]string, 0),
	}
	anno := sandbox.GetAnnotations()
	if anno == nil {
		return released, nil
	}
	if raw := anno[AnnoAllocReleasedKey]; raw != "" {
		err := json.Unmarshal([]byte(raw), released)
		if err != nil {
			return nil, err
		}
	}
	return released, nil
}

func (syncer *annoAllocationSyncer) SetReleased(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox, released *AllocationReleased) error {
	js, err := json.Marshal(released)
	if err != nil {
		return err
	}
	anno := sandbox.GetAnnotations()
	if anno == nil {
		anno = make(map[string]string)
	}
	anno[AnnoAllocReleasedKey] = string(js)
	sandbox.SetAnnotations(anno)

	needRemoveFinalizer := false
	// If the sandbox is being deleted and all allocated pods have been released,
	// remove the finalizer so the sandbox can be garbage collected.
	if !sandbox.DeletionTimestamp.IsZero() {
		allocation, err := syncer.GetAllocation(ctx, sandbox)
		if err != nil {
			return err
		}
		releasedSet := make(map[string]struct{}, len(released.Pods))
		for _, p := range released.Pods {
			releasedSet[p] = struct{}{}
		}
		allReleased := true
		for _, p := range allocation.Pods {
			if _, ok := releasedSet[p]; !ok {
				allReleased = false
				break
			}
		}
		if allReleased && controllerutil.ContainsFinalizer(sandbox, FinalizerPoolAllocation) {
			needRemoveFinalizer = true
			filtered := make([]string, 0, len(sandbox.GetFinalizers()))
			for _, f := range sandbox.GetFinalizers() {
				if f != FinalizerPoolAllocation {
					filtered = append(filtered, f)
				}
			}
			sandbox.SetFinalizers(filtered)
		}
	}

	meta := map[string]any{
		"annotations": map[string]string{
			AnnoAllocReleasedKey: string(js),
		},
	}
	if needRemoveFinalizer {
		meta["finalizers"] = sandbox.GetFinalizers()
	}
	patchData, err := json.Marshal(map[string]any{"metadata": meta})
	if err != nil {
		return err
	}
	obj := &sandboxv1alpha1.BatchSandbox{}
	obj.Name = sandbox.Name
	obj.Namespace = sandbox.Namespace
	return syncer.client.Patch(ctx, obj, client.RawPatch(types.MergePatchType, patchData))
}

type AllocSpec struct {
	// Sandboxes contains all BatchSandboxes to be scheduled in this round.
	Sandboxes []*sandboxv1alpha1.BatchSandbox
	Pool      *sandboxv1alpha1.Pool
	// Pods contains all candidate pods owned by the pool.
	Pods []*corev1.Pod
}

type Allocator interface {
	Schedule(ctx context.Context, spec *AllocSpec) (*algorithm.AllocAction, error)
	GetPoolAllocation(ctx context.Context, pool *sandboxv1alpha1.Pool) (map[string]string, error)
	ClearPoolAllocation(ctx context.Context, ns string, poolName string) error
	// ReleasePodsAllocation releases the in-memory allocation for the given pods directly,
	// without persisting to an annotation. Used for orphan pods whose sandbox no longer exists.
	ReleasePodsAllocation(ctx context.Context, ns string, poolName string, pods []string)
	SyncSandboxAllocation(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox, pods []string) error
	SyncSandboxReleased(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox, pods []string) error
	GetSandboxAllocation(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox) ([]string, error)
	GetSandboxReleased(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox) ([]string, error)
}

type defaultAllocator struct {
	store       AllocationStore
	syncer      AllocationSyncer
	client      client.Client
	algorithm   algorithm.Algorithm
	recoverOnce sync.Once
}

func NewDefaultAllocator(client client.Client) Allocator {
	return &defaultAllocator{
		store:     NewInMemoryAllocationStore(),
		syncer:    NewAnnoAllocationSyncer(client),
		client:    client,
		algorithm: &algorithm.PackedSchedule{},
	}
}

func (allocator *defaultAllocator) Schedule(ctx context.Context, spec *AllocSpec) (*algorithm.AllocAction, error) {
	log := logf.FromContext(ctx)
	log.Info("Schedule started", "pool", spec.Pool.Name, "totalPods", len(spec.Pods), "sandboxes", len(spec.Sandboxes))
	if err := allocator.checkRecovery(ctx); err != nil {
		return nil, err
	}

	// Fetch pool allocation once and reuse it for both stale-sandbox cleanup and available-pod filtering.
	// This avoids a double store read on every reconcile.
	podAllocation, err := allocator.GetPoolAllocation(ctx, spec.Pool)
	if err != nil {
		return nil, err
	}

	// GC + per-sandbox allocation requests: build SandboxRequests for all existing sandboxes and
	// append orphan entries for pods whose sandbox no longer exists (e.g. force-deleted).
	// Orphan entries carry PodSupplement=0 and ToRelease=orphan pods so the normal recycle path
	// handles them without any special-casing outside this function.
	// Terminating sandboxes are handled inside getSandboxRequest: they receive no new supplement and
	// all unreleased pods are queued for release.
	allRequest, err := allocator.getAllRequest(ctx, spec.Sandboxes, podAllocation)
	if err != nil {
		return nil, err
	}

	// Build available pod list using the already-fetched allocation to avoid an extra store read.
	availablePods, err := allocator.getAvailablePodsFromAlloc(ctx, podAllocation, spec.Pods)
	if err != nil {
		return nil, err
	}

	// Run the allocation algorithm.
	action := allocator.algorithm.Schedule(availablePods, allRequest)

	return action, nil
}

// getAllRequest builds per-sandbox allocation requests for all existing sandboxes and appends
// orphan entries for pods in podAllocation whose sandbox is no longer in the sandboxes list
// (e.g. force-deleted). Orphan entries carry PodSupplement=0 and ToRelease set to the orphan
// pods so the normal recycle path handles them without special-casing in the caller.
func (allocator *defaultAllocator) getAllRequest(ctx context.Context, sandboxes []*sandboxv1alpha1.BatchSandbox, podAllocation map[string]string) ([]*algorithm.SandboxRequest, error) {
	log := logf.FromContext(ctx)
	existingSandboxes := make(map[string]struct{}, len(sandboxes))
	allRequest := make([]*algorithm.SandboxRequest, 0, len(sandboxes))
	for _, sandbox := range sandboxes {
		existingSandboxes[sandbox.Name] = struct{}{}
		request, err := allocator.getSandboxRequest(ctx, sandbox)
		if err != nil {
			return nil, err
		}
		allRequest = append(allRequest, request)
	}
	orphanToRelease := make(map[string][]string)
	for podName, sandboxName := range podAllocation {
		if _, exists := existingSandboxes[sandboxName]; !exists {
			orphanToRelease[sandboxName] = append(orphanToRelease[sandboxName], podName)
		}
	}
	for sandboxName, pods := range orphanToRelease {
		log.Info("GC: queuing orphan pods for release", "sandbox", sandboxName, "pods", pods)
		allRequest = append(allRequest, &algorithm.SandboxRequest{
			SandboxName:   sandboxName,
			PodSupplement: 0,
			ToRelease:     pods,
		})
	}
	return allRequest, nil
}

func (allocator *defaultAllocator) getSandboxRequest(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox) (*algorithm.SandboxRequest, error) {
	log := logf.FromContext(ctx)
	allocated, err := allocator.GetSandboxAllocation(ctx, sandbox)
	if err != nil {
		return nil, err
	}
	released, err := allocator.GetSandboxReleased(ctx, sandbox)
	if err != nil {
		return nil, err
	}

	releasedSet := make(map[string]struct{}, len(released))
	for _, r := range released {
		releasedSet[r] = struct{}{}
	}

	// Terminating sandboxes should not receive new allocations.
	// Queue all unreleased allocated pods for release and set supplement to zero.
	if !sandbox.DeletionTimestamp.IsZero() {
		toRelease := make([]string, 0)
		for _, p := range allocated {
			if _, ok := releasedSet[p]; !ok {
				toRelease = append(toRelease, p)
			}
		}
		if len(toRelease) > 0 {
			log.Info("Queuing terminating sandbox pods for release", "sandbox", sandbox.Name, "pods", toRelease)
		}
		return &algorithm.SandboxRequest{
			SandboxName:   sandbox.Name,
			CurAllocation: allocated,
			CurReleased:   released,
			PodSupplement: 0,
			ToRelease:     toRelease,
		}, nil
	}

	release, err := allocator.getSandboxRelease(ctx, sandbox)
	if err != nil {
		return nil, err
	}

	toRelease := make([]string, 0)
	for _, r := range release {
		if _, exists := releasedSet[r]; !exists {
			toRelease = append(toRelease, r)
		}
	}

	replica := int32(0)
	if sandbox.Spec.Replicas != nil {
		replica = *sandbox.Spec.Replicas
	}

	supplement := int32(0)
	if replica-int32(len(allocated)) > 0 {
		supplement = replica - int32(len(allocated))
	}

	return &algorithm.SandboxRequest{
		SandboxName:   sandbox.Name,
		CurAllocation: allocated,
		CurReleased:   released,
		PodSupplement: supplement,
		ToRelease:     toRelease,
	}, nil
}

func (allocator *defaultAllocator) GetSandboxAllocation(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox) ([]string, error) {
	allocation, err := allocator.syncer.GetAllocation(ctx, sandbox)
	if err != nil {
		return nil, err
	}
	return allocation.Pods, nil
}

func (allocator *defaultAllocator) getSandboxRelease(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox) ([]string, error) {
	release, err := allocator.syncer.GetRelease(ctx, sandbox)
	if err != nil {
		return nil, err
	}
	return release.Pods, nil
}

func (allocator *defaultAllocator) GetSandboxReleased(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox) ([]string, error) {
	released, err := allocator.syncer.GetReleased(ctx, sandbox)
	if err != nil {
		return nil, err
	}
	return released.Pods, nil
}

func (allocator *defaultAllocator) GetPoolAllocation(ctx context.Context, pool *sandboxv1alpha1.Pool) (map[string]string, error) {
	alloc, err := allocator.store.GetAllocation(ctx, pool)
	if err != nil {
		return nil, err
	}
	if alloc == nil {
		return map[string]string{}, nil
	}
	return alloc.PodAllocation, nil
}

func (allocator *defaultAllocator) ClearPoolAllocation(ctx context.Context, ns string, poolName string) error {
	return allocator.store.ClearAllocation(ctx, ns, poolName)
}

func (allocator *defaultAllocator) ReleasePodsAllocation(ctx context.Context, ns string, poolName string, pods []string) {
	allocator.store.ReleaseAllocation(ctx, ns, poolName, pods)
}

// SyncSandboxAllocation updates the in-memory allocation store and then persists to the sandbox annotation.
// If annotation sync fails, the in-memory store is rolled back to the previous state to maintain consistency.
func (allocator *defaultAllocator) SyncSandboxAllocation(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox, pods []string) error {
	log := logf.FromContext(ctx)
	log.Info("Syncing sandbox allocation", "sandbox", sandbox.Name, "pods", pods)
	poolRef := sandbox.Spec.PoolRef

	// Snapshot the current in-memory state for rollback on failure.
	oldState, err := allocator.syncer.GetAllocation(ctx, sandbox)
	if err != nil {
		return fmt.Errorf("failed to get current sandbox allocation: %w", err)
	}

	// Phase 1: update in-memory store optimistically.
	allocator.store.UpdateAllocation(ctx, sandbox.Namespace, poolRef, sandbox.Name, pods)

	// Phase 2: persist to sandbox annotation.
	allocation := &SandboxAllocation{Pods: pods}
	if err := allocator.syncer.SetAllocation(ctx, sandbox, allocation); err != nil {
		// Rollback in-memory store to the previous state.
		log.Error(err, "Rollback sandbox allocation", "sandbox", sandbox.Name, "pods", oldState.Pods)
		allocator.store.UpdateAllocation(ctx, sandbox.Namespace, poolRef, sandbox.Name, oldState.Pods)
		return err
	}
	return nil
}

// SyncSandboxReleased persists the released state to the sandbox annotation and then releases from the in-memory store.
// Annotation must succeed before the pods are removed from the in-memory store to prevent pods from being
// re-allocated before the release is durably committed.
func (allocator *defaultAllocator) SyncSandboxReleased(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox, pods []string) error {
	log := logf.FromContext(ctx)
	log.Info("Syncing sandbox released", "sandbox", sandbox.Name, "pods", pods)
	poolRef := sandbox.Spec.PoolRef

	// Phase 1: persist to sandbox annotation.
	released := &AllocationReleased{Pods: pods}
	if err := allocator.syncer.SetReleased(ctx, sandbox, released); err != nil {
		log.Error(err, "Failed to sync sandbox released", "sandbox", sandbox.Name, "pods", pods)
		return err
	}

	// Phase 2: release from in-memory store only after the annotation is durably committed.
	allocator.store.ReleaseAllocation(ctx, sandbox.Namespace, poolRef, pods)
	return nil
}

// checkRecovery runs the one-time state recovery. If recovery fails the process
// is terminated via os.Exit(1) because the allocator cannot operate with an
// inconsistent in-memory state. The error return is kept for interface compatibility
// but will never actually be returned.
func (allocator *defaultAllocator) checkRecovery(ctx context.Context) error {
	allocator.recoverOnce.Do(func() {
		log := logf.FromContext(ctx)
		if err := allocator.store.Recover(ctx, allocator.client); err != nil {
			log.Error(err, "Fatal: allocator state recovery failed, exiting")
			os.Exit(1)
		}
	})
	return nil
}

func (allocator *defaultAllocator) getAvailablePodsFromAlloc(_ context.Context, podAllocation map[string]string, pods []*corev1.Pod) ([]string, error) {
	availablePods := make([]string, 0)
	for _, pod := range pods {
		if _, ok := podAllocation[pod.Name]; ok {
			continue
		}
		if !utils.IsPodReady(pod) {
			continue
		}
		availablePods = append(availablePods, pod.Name)
	}
	return availablePods, nil
}