iop/packages/go/streamgate/runtime.go

package streamgate

import (
	"errors"
)

const (
	// DefaultMaxEvidenceRunes is the default Unicode rune threshold for evidence holding.
	DefaultMaxEvidenceRunes = 500

	// DefaultMaxBufferRunes is the default maximum rune buffer size.
	DefaultMaxBufferRunes = 4096

	// DefaultMaxIngressSnapshotBytes is the default byte limit for ingress body snapshots (16 MiB).
	DefaultMaxIngressSnapshotBytes int64 = 16 * 1024 * 1024

	// DefaultMaxRecoveryAttemptsTotal is the default total fault recovery attempt limit.
	DefaultMaxRecoveryAttemptsTotal = 3
)

// RuntimeOptions captures immutable configuration parameters for a streamgate request runtime.
type RuntimeOptions struct {
	maxEvidenceRunes         int
	maxBufferRunes           int
	maxIngressSnapshotBytes  int64
	maxRecoveryAttemptsTotal int
	gateOptions              GateCoordinatorOptions
	recoveryOptions          RecoveryCoordinatorOptions
}

// DefaultRuntimeOptions returns RuntimeOptions populated with default settings.
func DefaultRuntimeOptions() RuntimeOptions {
	opts, _ := NewRuntimeOptions(
		DefaultMaxEvidenceRunes,
		DefaultMaxBufferRunes,
		DefaultMaxIngressSnapshotBytes,
		DefaultMaxRecoveryAttemptsTotal,
		GateCoordinatorOptions{},
		RecoveryCoordinatorOptions{},
	)
	return opts
}

// NewRuntimeOptions constructs validated RuntimeOptions.
func NewRuntimeOptions(
	maxEvidenceRunes, maxBufferRunes int,
	maxIngressBytes int64,
	maxRecoveryAttempts int,
	gateOpts GateCoordinatorOptions,
	recoveryOpts RecoveryCoordinatorOptions,
) (RuntimeOptions, error) {
	if maxEvidenceRunes <= 0 {
		return RuntimeOptions{}, errors.New("streamgate: runtime options max evidence runes must be positive")
	}
	if maxBufferRunes < maxEvidenceRunes {
		return RuntimeOptions{}, errors.New("streamgate: runtime options max buffer runes must be >= max evidence runes")
	}
	if maxIngressBytes <= 0 {
		return RuntimeOptions{}, errors.New("streamgate: runtime options max ingress snapshot bytes must be positive")
	}
	if maxRecoveryAttempts < 0 || maxRecoveryAttempts > DefaultMaxRecoveryAttemptsTotal {
		return RuntimeOptions{}, errors.New("streamgate: runtime options max recovery attempts total must be between 0 and 3")
	}
	opts := RuntimeOptions{
		maxEvidenceRunes:         maxEvidenceRunes,
		maxBufferRunes:           maxBufferRunes,
		maxIngressSnapshotBytes:  maxIngressBytes,
		maxRecoveryAttemptsTotal: maxRecoveryAttempts,
		gateOptions:              gateOpts,
		recoveryOptions:          recoveryOpts,
	}
	if err := opts.Validate(); err != nil {
		return RuntimeOptions{}, err
	}
	return opts, nil
}

// Validate returns nil when RuntimeOptions is in a consistent state.
func (o RuntimeOptions) Validate() error {
	if o.maxEvidenceRunes <= 0 {
		return errors.New("streamgate: runtime options max evidence runes must be positive")
	}
	if o.maxBufferRunes < o.maxEvidenceRunes {
		return errors.New("streamgate: runtime options max buffer runes must be >= max evidence runes")
	}
	if o.maxIngressSnapshotBytes <= 0 || o.maxIngressSnapshotBytes > DefaultMaxIngressSnapshotBytes {
		return errors.New("streamgate: runtime options max ingress snapshot bytes must be between 1 and 16777216")
	}
	if o.maxRecoveryAttemptsTotal < 0 || o.maxRecoveryAttemptsTotal > DefaultMaxRecoveryAttemptsTotal {
		return errors.New("streamgate: runtime options max recovery attempts total must be between 0 and 3")
	}
	return nil
}

// MaxEvidenceRunes returns the max evidence runes setting.
func (o RuntimeOptions) MaxEvidenceRunes() int { return o.maxEvidenceRunes }

// MaxBufferRunes returns the max buffer runes setting.
func (o RuntimeOptions) MaxBufferRunes() int { return o.maxBufferRunes }

// MaxIngressSnapshotBytes returns the max ingress snapshot bytes setting.
func (o RuntimeOptions) MaxIngressSnapshotBytes() int64 { return o.maxIngressSnapshotBytes }

// MaxRecoveryAttemptsTotal returns the total recovery attempts limit.
func (o RuntimeOptions) MaxRecoveryAttemptsTotal() int { return o.maxRecoveryAttemptsTotal }

// GateOptions returns the gate coordinator options.
func (o RuntimeOptions) GateOptions() GateCoordinatorOptions { return o.gateOptions }

// RecoveryOptions returns the recovery coordinator options.
func (o RuntimeOptions) RecoveryOptions() RecoveryCoordinatorOptions { return o.recoveryOptions }

// RequestRuntimeSnapshot binds request-local state, host seams, filter registry snapshot,
// and runtime options into an immutable contract to be consumed by request-local owners.
type RequestRuntimeSnapshot struct {
	requestID          StableToken
	configGen          StableToken
	environment        string
	endpoint           string
	family             string
	options            RuntimeOptions
	registrySnapshot   FilterRegistrySnapshot
	ingressSnapshot    *IngressSnapshot
	requestSnapshotRef RecoveryRequestSnapshotRef
	dispatcher         AttemptDispatcher
	rebuilder          RequestRebuilder
	preparer           RecoveryPlanPreparer
	sink               ReleaseSink
}

// NewRequestRuntimeSnapshot constructs a validated RequestRuntimeSnapshot.
func NewRequestRuntimeSnapshot(
	requestID, configGen, environment, endpoint, family string,
	options RuntimeOptions,
	registrySnapshot FilterRegistrySnapshot,
	ingressSnapshot *IngressSnapshot,
	requestSnapshotRef RecoveryRequestSnapshotRef,
	dispatcher AttemptDispatcher,
	rebuilder RequestRebuilder,
	preparer RecoveryPlanPreparer,
	sink ReleaseSink,
) (RequestRuntimeSnapshot, error) {
	reqIDToken, err := NewStableTokenRequired("requestID", requestID)
	if err != nil {
		return RequestRuntimeSnapshot{}, err
	}
	cfgGenToken, err := NewStableTokenRequired("configGen", configGen)
	if err != nil {
		return RequestRuntimeSnapshot{}, err
	}
	if err := options.Validate(); err != nil {
		return RequestRuntimeSnapshot{}, err
	}
	if registrySnapshot.Generation() == "" {
		return RequestRuntimeSnapshot{}, errors.New("streamgate: request runtime snapshot registry snapshot generation is required")
	}
	if dispatcher == nil {
		return RequestRuntimeSnapshot{}, errors.New("streamgate: request runtime snapshot dispatcher is required")
	}
	if rebuilder == nil {
		return RequestRuntimeSnapshot{}, errors.New("streamgate: request runtime snapshot rebuilder is required")
	}
	if sink == nil {
		return RequestRuntimeSnapshot{}, errors.New("streamgate: request runtime snapshot release sink is required")
	}
	if err := requestSnapshotRef.Validate(); err != nil {
		return RequestRuntimeSnapshot{}, err
	}

	snap := RequestRuntimeSnapshot{
		requestID:          reqIDToken,
		configGen:          cfgGenToken,
		environment:        environment,
		endpoint:           endpoint,
		family:             family,
		options:            options,
		registrySnapshot:   registrySnapshot,
		ingressSnapshot:    ingressSnapshot,
		requestSnapshotRef: requestSnapshotRef,
		dispatcher:         dispatcher,
		rebuilder:          rebuilder,
		preparer:           preparer,
		sink:               sink,
	}

	if err := snap.Validate(); err != nil {
		return RequestRuntimeSnapshot{}, err
	}
	return snap, nil
}

// Validate returns nil when the RequestRuntimeSnapshot is consistent.
func (r RequestRuntimeSnapshot) Validate() error {
	if err := validateStableTokenRequired("requestID", r.requestID.value); err != nil {
		return err
	}
	if err := validateStableTokenRequired("configGen", r.configGen.value); err != nil {
		return err
	}
	if err := r.options.Validate(); err != nil {
		return err
	}
	registryGeneration := r.registrySnapshot.Generation()
	if registryGeneration == "" {
		return errors.New("streamgate: request runtime snapshot registry snapshot generation is required")
	}
	if registryGeneration != r.configGen.String() {
		return errors.New("streamgate: request runtime snapshot config and registry generations must match")
	}
	if r.dispatcher == nil {
		return errors.New("streamgate: request runtime snapshot dispatcher is required")
	}
	if r.rebuilder == nil {
		return errors.New("streamgate: request runtime snapshot rebuilder is required")
	}
	if r.sink == nil {
		return errors.New("streamgate: request runtime snapshot release sink is required")
	}
	if err := r.requestSnapshotRef.Validate(); err != nil {
		return err
	}
	return nil
}

// RequestID returns the stable request identifier string.
func (r RequestRuntimeSnapshot) RequestID() string { return r.requestID.value }

// ConfigGeneration returns the configuration generation string.
func (r RequestRuntimeSnapshot) ConfigGeneration() string { return r.configGen.value }

// Environment returns the environment label.
func (r RequestRuntimeSnapshot) Environment() string { return r.environment }

// Endpoint returns the endpoint label.
func (r RequestRuntimeSnapshot) Endpoint() string { return r.endpoint }

// Family returns the model family label.
func (r RequestRuntimeSnapshot) Family() string { return r.family }

// Options returns the runtime options.
func (r RequestRuntimeSnapshot) Options() RuntimeOptions { return r.options }

// RegistrySnapshot returns the filter registry snapshot.
func (r RequestRuntimeSnapshot) RegistrySnapshot() FilterRegistrySnapshot {
	return r.registrySnapshot
}

// IngressSnapshot returns the ingress snapshot pointer, or nil if unset.
func (r RequestRuntimeSnapshot) IngressSnapshot() *IngressSnapshot {
	return r.ingressSnapshot
}

// RequestSnapshotRef returns the recovery request snapshot reference.
func (r RequestRuntimeSnapshot) RequestSnapshotRef() RecoveryRequestSnapshotRef {
	return r.requestSnapshotRef
}

// Dispatcher returns the host attempt dispatcher.
func (r RequestRuntimeSnapshot) Dispatcher() AttemptDispatcher { return r.dispatcher }

// Rebuilder returns the host request rebuilder.
func (r RequestRuntimeSnapshot) Rebuilder() RequestRebuilder { return r.rebuilder }

// Preparer returns the optional host recovery plan preparer.
func (r RequestRuntimeSnapshot) Preparer() RecoveryPlanPreparer { return r.preparer }

// Sink returns the host release sink.
func (r RequestRuntimeSnapshot) Sink() ReleaseSink { return r.sink }

// BuildAttemptFilterContext constructs a FilterContext bound to this request snapshot.
func (r RequestRuntimeSnapshot) BuildAttemptFilterContext(
	attemptID string,
	target AttemptTarget,
	epochID uint64,
	commitState CommitState,
) (FilterContext, error) {
	builder := NewFilterContextBuilder(r.configGen.value, attemptID).
		SetEnvironment(r.environment).
		SetEndpoint(r.endpoint).
		SetFamily(r.family).
		SetModelGroup(target.ModelGroup()).
		SetActualModel(target.Model()).
		SetActualProvider(target.Provider()).
		SetExecutionPath(target.ExecutionPath()).
		SetCommitState(commitState).
		SetStableCorrelation(r.requestID.value)

	if epochID > 0 {
		var err error
		builder, err = builder.SetEpoch(epochID)
		if err != nil {
			return FilterContext{}, err
		}
	}
	return builder.Build()
}