proxy-pool/internal/controller/routing/sequential.go

// Package routing coordinates Controller-side automatic Sequential switching.
// It consumes bounded Provider result notifications and keeps the actual state
// transition in the existing management-store compare-and-swap boundary.
package routing

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"reflect"
	"strconv"
	"sync"
	"time"

	"proxy-pool/internal/config"
	"proxy-pool/internal/controller/provider"
	"proxy-pool/internal/domain/adminstate"
	"proxy-pool/internal/domain/upstream"
)

var ErrInvalidSequentialCoordinator = errors.New("invalid sequential routing coordinator")

const (
	defaultPollInterval = time.Second
	autoSwitchActor     = "proxy-controller"
	autoSwitchReason    = "consecutive empty provider fetches reached routing threshold"
)

// ConfigurationSource supplies one immutable configuration and its matching
// authoritative management revision.
type ConfigurationSource interface {
	Snapshot() (*config.Config, uint64)
}

// StateStore contains the only mutation boundary for routing current-upstream
// state. Implementations enforce ExpectedCurrent as a compare-and-swap fence.
type StateStore interface {
	Snapshot(context.Context) (adminstate.Snapshot, error)
	SwitchRouting(context.Context, adminstate.SwitchRoutingCommand) (adminstate.MutationResult, error)
}

type SnapshotRefreshNotifier interface {
	NotifySnapshotRefresh()
}

type Options struct {
	Now          func() time.Time
	PollInterval time.Duration
}

type TickResult struct {
	Switched int
}

// SequentialCoordinator observes Provider fetch outcomes without blocking the
// fetch path. Tick serializes bounded scans and only records a generation as
// processed after it has made, or intentionally declined, a decision.
type SequentialCoordinator struct {
	configuration ConfigurationSource
	state         StateStore
	stats         provider.StatsReader
	refresh       SnapshotRefreshNotifier
	now           func() time.Time
	pollInterval  time.Duration
	notify        chan struct{}

	mu        sync.Mutex
	processed map[string]uint64
}

var _ provider.ResultObserver = (*SequentialCoordinator)(nil)

func NewSequentialCoordinator(
	configuration ConfigurationSource,
	state StateStore,
	stats provider.StatsReader,
	refresh SnapshotRefreshNotifier,
	options Options,
) (*SequentialCoordinator, error) {
	if nilInterface(configuration) || nilInterface(state) || nilInterface(stats) || options.Now == nil {
		return nil, ErrInvalidSequentialCoordinator
	}
	if options.PollInterval == 0 {
		options.PollInterval = defaultPollInterval
	}
	if options.PollInterval <= 0 {
		return nil, ErrInvalidSequentialCoordinator
	}
	return &SequentialCoordinator{
		configuration: configuration,
		state:         state,
		stats:         stats,
		refresh:       refresh,
		now:           options.Now,
		pollInterval:  options.PollInterval,
		notify:        make(chan struct{}, 1),
		processed:     make(map[string]uint64),
	}, nil
}

// ObserveProviderResult deliberately performs no store calls. Every validated
// result can affect an empty episode, but one buffered notification is enough
// because Tick reads the latest bounded Stats snapshot.
func (coordinator *SequentialCoordinator) ObserveProviderResult(result provider.Result) {
	if coordinator == nil || !validResultClass(result.Class) || result.UpstreamID == "" {
		return
	}
	select {
	case coordinator.notify <- struct{}{}:
	default:
	}
}

func (coordinator *SequentialCoordinator) Run(ctx context.Context) error {
	if coordinator == nil || ctx == nil || coordinator.pollInterval <= 0 || coordinator.now == nil || coordinator.notify == nil {
		return ErrInvalidSequentialCoordinator
	}
	ticker := time.NewTicker(coordinator.pollInterval)
	defer ticker.Stop()
	for {
		// A temporary management-store failure must not terminate Provider
		// reconciliation. The retained empty generation is evaluated on retry.
		_, _ = coordinator.Tick(ctx)
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-coordinator.notify:
		case <-ticker.C:
		}
	}
}

func (coordinator *SequentialCoordinator) Tick(ctx context.Context) (TickResult, error) {
	if coordinator == nil || ctx == nil || nilInterface(coordinator.configuration) || nilInterface(coordinator.state) ||
		nilInterface(coordinator.stats) || coordinator.now == nil {
		return TickResult{}, ErrInvalidSequentialCoordinator
	}
	if err := ctx.Err(); err != nil {
		return TickResult{}, err
	}
	coordinator.mu.Lock()
	defer coordinator.mu.Unlock()

	configuration, revision := coordinator.configuration.Snapshot()
	if configuration == nil || revision == 0 {
		return TickResult{}, ErrInvalidSequentialCoordinator
	}
	snapshot, err := coordinator.state.Snapshot(ctx)
	if err != nil {
		return TickResult{}, err
	}
	if snapshot.Config == nil || snapshot.Config.Revision != revision {
		return TickResult{}, ErrInvalidSequentialCoordinator
	}

	upstreams := upstreamStates(snapshot.Upstreams)
	routings := routingStates(snapshot.Routings)
	result := TickResult{}
	for _, route := range configuration.Routing {
		if !route.Enabled || route.Strategy.Type != "sequential" || route.Strategy.SwitchAfterEmptyFetch <= 0 {
			continue
		}
		state, exists := routings[route.Name]
		if !exists || !state.Enabled || !upstreamEnabled(configuration, upstreams, state.CurrentUpstream) {
			continue
		}
		read := coordinator.stats.ReadProviderStats([]string{state.CurrentUpstream})
		if len(read) != 1 {
			return result, ErrInvalidSequentialCoordinator
		}
		stats := read[0]
		if stats.UpstreamID != state.CurrentUpstream || stats.ConsecutiveEmptyFetch < int64(route.Strategy.SwitchAfterEmptyFetch) ||
			stats.EmptyGeneration == 0 {
			coordinator.clearProcessed(route.Name, state.CurrentUpstream)
			continue
		}
		key := processedKey(route.Name, state.CurrentUpstream)
		if coordinator.processed[key] == stats.EmptyGeneration {
			continue
		}
		target, switchable := nextUpstream(route, state.CurrentUpstream, upstreams, configuration)
		if !switchable {
			coordinator.processed[key] = stats.EmptyGeneration
			continue
		}
		mutation, mutateErr := coordinator.state.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{
			RequestID: requestID(route.Name, state.CurrentUpstream, stats.EmptyGeneration),
			Actor:     adminstate.Actor{ID: autoSwitchActor}, OccurredAt: coordinator.now().UTC(),
			Name: route.Name, ExpectedCurrent: state.CurrentUpstream, Target: target, Reason: autoSwitchReason,
		})
		if mutateErr != nil {
			if errors.Is(mutateErr, adminstate.ErrConflict) {
				coordinator.processed[key] = stats.EmptyGeneration
				continue
			}
			return result, mutateErr
		}
		coordinator.processed[key] = stats.EmptyGeneration
		if mutation.Changed {
			result.Switched++
			if coordinator.refresh != nil {
				coordinator.refresh.NotifySnapshotRefresh()
			}
		}
	}
	return result, nil
}

func nextUpstream(
	route config.Routing,
	current string,
	states map[string]adminstate.UpstreamState,
	configuration *config.Config,
) (string, bool) {
	eligible := make([]string, 0, len(route.Upstreams))
	currentIndex := -1
	for _, upstream := range route.Upstreams {
		if !upstreamEnabled(configuration, states, upstream) {
			continue
		}
		if upstream == current {
			currentIndex = len(eligible)
		}
		eligible = append(eligible, upstream)
	}
	if currentIndex < 0 || len(eligible) < 2 {
		return "", false
	}
	if currentIndex+1 < len(eligible) {
		return eligible[currentIndex+1], true
	}
	switch route.Strategy.EndBehavior {
	case "loop":
		return eligible[0], eligible[0] != current
	case "stayLast", "", "stop":
		return "", false
	default:
		return "", false
	}
}

func upstreamStates(values []adminstate.UpstreamState) map[string]adminstate.UpstreamState {
	result := make(map[string]adminstate.UpstreamState, len(values))
	for _, value := range values {
		result[value.Name] = value
	}
	return result
}

func routingStates(values []adminstate.RoutingState) map[string]adminstate.RoutingState {
	result := make(map[string]adminstate.RoutingState, len(values))
	for _, value := range values {
		result[value.Name] = value
	}
	return result
}

func upstreamEnabled(configuration *config.Config, states map[string]adminstate.UpstreamState, name string) bool {
	configured, exists := configuration.Upstreams[name]
	if !exists || !configured.Enabled {
		return false
	}
	state, exists := states[name]
	return exists && state.Enabled
}

func (coordinator *SequentialCoordinator) clearProcessed(routeName, upstream string) {
	delete(coordinator.processed, processedKey(routeName, upstream))
}

func processedKey(routeName, upstream string) string {
	return routeName + "\x00" + upstream
}

func requestID(routeName, upstream string, generation uint64) string {
	payload := routeName + "\x00" + upstream + "\x00" + strconv.FormatUint(generation, 10)
	digest := sha256.Sum256([]byte(payload))
	return "auto-sequential-" + hex.EncodeToString(digest[:16])
}

func validResultClass(class upstream.FetchClass) bool {
	switch class {
	case upstream.FetchValid, upstream.FetchEmpty, upstream.FetchDuplicateOnly, upstream.FetchError:
		return true
	default:
		return false
	}
}

func nilInterface(value any) bool {
	if value == nil {
		return true
	}
	reflected := reflect.ValueOf(value)
	switch reflected.Kind() {
	case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice:
		return reflected.IsNil()
	default:
		return false
	}
}