iop/packages/go/streamgate/parallel_evaluation.go

package streamgate

import (
	"context"
	"errors"
	"sync"
	"time"
)

// EpochArbiter is invoked by the coordinator after all filter outcomes for a
// single epoch are collected into a stable EvaluationSet. The coordinator
// calls Arbitrate at most once per epoch and only when the complete outcome
// set is fully assembled.
type EpochArbiter interface {
	Arbitrate(context.Context, EvidenceBatch, EvaluationSet) (ArbitrationResult, error)
}

// GateCoordinatorOptions configures the coordinator's ingress behavior.
//
// MaxQueuedEpochs bounds the number of pending epoch jobs. Only the actively
// executing epoch is processed by the single worker; up to MaxQueuedEpochs
// jobs may queue behind it. Submitters beyond this bound block on send.
type GateCoordinatorOptions struct {
	// MaxQueuedEpochs bounds pending epoch jobs. Zero means no queueing —
	// only the actively executing epoch is allowed.
	MaxQueuedEpochs int

	// ObservationSink receives filter observation events during epoch evaluation.
	// If nil, NoopObservationSink is used by default.
	ObservationSink ObservationSink
}

// observationCtxKey is the context key for carrying observation correlation.
type observationCtxKey struct{}

// ObservationContext carries request-local correlation metadata and sequencer into GateCoordinator.
type ObservationContext struct {
	Sequencer         *ObservationSequencer
	StableCorrelation string
	ConfigGeneration  string
	AttemptID         string
	AttemptTarget     ObservationAttemptTarget
	EpochID           uint64
}

// WithObservationContext returns a child context carrying ObservationContext.
func WithObservationContext(ctx context.Context, obsCtx ObservationContext) context.Context {
	return context.WithValue(ctx, observationCtxKey{}, obsCtx)
}

// ObservationContextFromContext retrieves ObservationContext from context if present.
func ObservationContextFromContext(ctx context.Context) (ObservationContext, bool) {
	v, ok := ctx.Value(observationCtxKey{}).(ObservationContext)
	return v, ok
}

// GateCoordinator drives immutable batch evaluation with an all-complete
// outcome barrier, single-flight worker processing, bounded backpressure, and
// structured cancellation.
//
// It owns exactly one worker goroutine that serializes epoch execution over
// a bounded jobs channel. Caller-supplied contexts are propagated into filter
// child contexts; filter evaluation timeouts are applied per filter. Raw
// evaluator errors and deadline overruns are normalized to stable, sanitized
// outcome codes with no raw error payload preserved.
type GateCoordinator struct {
	hostCtx    context.Context
	arbiter    EpochArbiter
	jobs       chan epochJob
	closedCh   chan struct{}
	closeOnce  sync.Once
	closedOnce sync.Once
	wg         sync.WaitGroup
	sequencer  *ObservationSequencer
}

type epochJob struct {
	ctx     context.Context
	batch   EvidenceBatch
	filters []EpochFilter
	reply   chan<- epochResult
}

type epochResult struct {
	set    EvaluationSet
	result ArbitrationResult
	err    error
}

// NewGateCoordinator constructs a coordinator.
//
// hostCtx drives the worker's lifecycle: when hostCtx is cancelled the worker
// stops accepting work, all in-flight epochs terminate without invoking the
// arbiter, and any queued epochs are completed with a host-cancel error.
//
// arbiter must not be nil. MaxQueuedEpochs must be non-negative.
func NewGateCoordinator(
	hostCtx context.Context,
	options GateCoordinatorOptions,
	arbiter EpochArbiter,
) (*GateCoordinator, error) {
	if arbiter == nil {
		return nil, errors.New("streamgate: new gate coordinator arbiter is required")
	}
	if options.MaxQueuedEpochs < 0 {
		return nil, errors.New("streamgate: new gate coordinator max queued epochs must be non-negative")
	}

	c := &GateCoordinator{
		hostCtx:   hostCtx,
		arbiter:   arbiter,
		jobs:      make(chan epochJob, options.MaxQueuedEpochs),
		closedCh:  make(chan struct{}),
		sequencer: NewObservationSequencer(options.ObservationSink, nil),
	}
	c.wg.Add(1)
	go c.worker()
	return c, nil
}

// SetObservationSequencer sets or overrides the coordinator's ObservationSequencer.
func (c *GateCoordinator) SetObservationSequencer(seq *ObservationSequencer) {
	if c != nil {
		c.sequencer = seq
	}
}

// worker runs exactly one epoch at a time, processing accepted jobs in FIFO
// order over the bounded jobs channel.
func (c *GateCoordinator) worker() {
	defer c.wg.Done()

	for {
		select {
		case job, ok := <-c.jobs:
			if !ok {
				return
			}
			c.runJob(job)
		case <-c.hostCtx.Done():
			// Close closedCh to unblock any submitters waiting on send.
			c.closedOnce.Do(func() { close(c.closedCh) })
			// Drain buffered jobs with a host-cancel error so their
			// submitters' reply channels are not left hanging.
			for {
				select {
				case job := <-c.jobs:
					select {
					case job.reply <- epochResult{err: c.hostCtx.Err()}:
					default:
					}
				default:
					return
				}
			}
		}
	}
}

func (c *GateCoordinator) runJob(job epochJob) {
	// Skip cancelled jobs without invoking evaluator or arbiter.
	if job.ctx.Err() != nil {
		select {
		case job.reply <- epochResult{err: job.ctx.Err()}:
		default:
		}
		return
	}

	set, res, err := c.Evaluate(job.ctx, job.batch, job.filters)
	select {
	case job.reply <- epochResult{set: set, result: res, err: err}:
	default:
	}
}

// Evaluate fans out ready filters for a single epoch, collects all outcomes
// through an all-complete barrier, and invokes the arbiter exactly once when
// the stable EvaluationSet is ready.
//
// Non-ready filters contribute pre-evaluation outcomes via NormalizeOutcome();
// ready filters are dispatched to goroutines with per-filter timeouts and the
// raw evaluator error is normalized to filter_evaluation_error or
// filter_evaluation_deadline codes. The caller/host context is propagated
// into every filter child context; if it is cancelled before the complete set
// is assembled the arbiter is NOT invoked.
func (c *GateCoordinator) Evaluate(ctx context.Context, batch EvidenceBatch, filters []EpochFilter) (EvaluationSet, ArbitrationResult, error) {
	if err := batch.Validate(); err != nil {
		return EvaluationSet{}, ArbitrationResult{}, errors.New("streamgate: evaluate batch: " + err.Error())
	}

	// Snapshot capturedAt so that epoch-level consistency can be verified by
	// downstream consumers/fixtures without exposing mutable batch internals.
	_ = batch.CapturedAt()

	// Partition filters into ready (subscribed + triggerReady) and non-ready.
	type filterWork struct {
		filter    EpochFilter
		outcomeCh chan FilterOutcome
	}

	var ready []filterWork
	var nonReady []EpochFilter

	for _, f := range filters {
		if f.EvaluatedForEpoch() {
			ch := make(chan FilterOutcome, 1)
			ready = append(ready, filterWork{filter: f, outcomeCh: ch})
		} else {
			nonReady = append(nonReady, f)
		}
	}

	if ctx.Err() != nil {
		return EvaluationSet{}, ArbitrationResult{}, ctx.Err()
	}

	seq := c.sequencer
	var correlation, configGen, attemptID string
	var attemptTarget ObservationAttemptTarget
	var epochID uint64

	if obsCtx, ok := ObservationContextFromContext(ctx); ok {
		if obsCtx.Sequencer != nil {
			seq = obsCtx.Sequencer
		}
		correlation = obsCtx.StableCorrelation
		configGen = obsCtx.ConfigGeneration
		attemptID = obsCtx.AttemptID
		attemptTarget = obsCtx.AttemptTarget
		epochID = obsCtx.EpochID
	}
	if correlation == "" {
		correlation = "default"
	}
	if configGen == "" {
		configGen = "default"
	}
	if attemptID == "" {
		attemptID = "default"
	}
	if err := attemptTarget.Validate(); err != nil {
		attemptTarget, _ = NewObservationAttemptTarget("default", "default", "default", "default")
	}

	// Emit filter_evaluation_started observations for ready filters.
	if seq != nil && epochID > 0 {
		for _, w := range ready {
			attr, _ := NewObservationAttribution("default", w.filter.ID(), "default")
			_, _ = seq.Emit(ctx, FilterObservationInput{
				Kind:              ObservationKindFilterEvaluationStarted,
				StableCorrelation: correlation,
				ConfigGeneration:  configGen,
				AttemptID:         attemptID,
				AttemptTarget:     attemptTarget,
				EpochID:           epochID,
				CommitState:       batch.CommitState(),
				Attribution:       &attr,
				OccurredAt:        time.Now(),
			})
		}
	}

	// Fan out ready filters into goroutines with per-filter timeouts.
	for _, w := range ready {
		go func(w filterWork) {
			defer close(w.outcomeCh)

			filterCtx, cancel := context.WithTimeout(ctx, w.filter.EvaluationTimeout())
			defer cancel()

			decision, evalErr := w.filter.Evaluator().Evaluate(filterCtx, batch)
			if filterCtx.Err() != nil {
				oc, _ := NewFilterOutcomeEvaluationError("filter_evaluation_deadline")
				w.outcomeCh <- oc
				return
			}
			if evalErr != nil {
				oc, _ := NewFilterOutcomeEvaluationError("filter_evaluation_error")
				w.outcomeCh <- oc
				return
			}
			oc, convErr := NewFilterOutcomeEvaluated(decision)
			if convErr != nil {
				oc, _ = NewFilterOutcomeEvaluationError("filter_evaluation_error")
				w.outcomeCh <- oc
				return
			}
			w.outcomeCh <- oc
		}(w)
	}

	// Collect all ready filter outcomes through the all-complete barrier.
	readyOutcomes := make([]FilterOutcome, 0, len(ready))
	for _, w := range ready {
		select {
		case oc := <-w.outcomeCh:
			readyOutcomes = append(readyOutcomes, oc)
		case <-ctx.Done():
			return EvaluationSet{}, ArbitrationResult{}, ctx.Err()
		case <-c.hostCtx.Done():
			return EvaluationSet{}, ArbitrationResult{}, c.hostCtx.Err()
		}
	}

	// Emit filter_evaluated observations for ready filters.
	if seq != nil && epochID > 0 {
		for i, w := range ready {
			oc := readyOutcomes[i]
			consumerID := "default"
			filterID := w.filter.ID()
			ruleID := "default"
			var decKind FilterDecisionKind
			var evi *SanitizedEvidence

			if oc.Kind() == FilterOutcomeKindEvaluated && oc.Decision() != nil {
				dec := oc.Decision()
				if dec.ConsumerID() != "" {
					consumerID = dec.ConsumerID()
				}
				if dec.FilterID() != "" {
					filterID = dec.FilterID()
				}
				if dec.RuleID() != "" {
					ruleID = dec.RuleID()
				}
				decKind = dec.Kind()
				e := dec.Evidence()
				if e.Count() > 0 || e.Fingerprint() != (FixedFingerprint{}) || e.DescriptorCode() != "" {
					evi = &e
				}
			}

			attr, _ := NewObservationAttribution(consumerID, filterID, ruleID)
			pol, _ := NewObservationDecisionPolicy(oc.Kind(), decKind, w.filter.EnforcedAs())

			_, _ = seq.Emit(ctx, FilterObservationInput{
				Kind:              ObservationKindFilterEvaluated,
				StableCorrelation: correlation,
				ConfigGeneration:  configGen,
				AttemptID:         attemptID,
				AttemptTarget:     attemptTarget,
				EpochID:           epochID,
				CommitState:       batch.CommitState(),
				Attribution:       &attr,
				DecisionPolicy:    &pol,
				Evidence:          evi,
				OccurredAt:        time.Now(),
			})
		}
	}

	// Build the stable complete outcome set.
	var outcomes []EpochFilterOutcome

	// Non-ready filters contribute pre-evaluation outcomes.
	for _, f := range nonReady {
		oc := f.NormalizeOutcome()
		if oc.Kind() != "" {
			o, err := NewEpochFilterOutcome(f.ID(), f.Priority(), oc, f.EnforcedAs())
			if err != nil {
				return EvaluationSet{}, ArbitrationResult{}, errors.New("streamgate: evaluate non-ready outcome: " + err.Error())
			}
			outcomes = append(outcomes, o)
		}
	}

	// Ready filters contribute evaluated outcomes.
	for i, w := range ready {
		o, err := NewEpochFilterOutcome(w.filter.ID(), w.filter.Priority(), readyOutcomes[i], w.filter.EnforcedAs())
		if err != nil {
			return EvaluationSet{}, ArbitrationResult{}, errors.New("streamgate: evaluate ready outcome: " + err.Error())
		}
		outcomes = append(outcomes, o)
	}

	// Construct the EvaluationSet (sorts by filter id ascending).
	set, err := NewEvaluationSet(outcomes)
	if err != nil {
		return EvaluationSet{}, ArbitrationResult{}, errors.New("streamgate: evaluate set: " + err.Error())
	}

	// Invoke the arbiter exactly once, only after the complete set is ready.
	arbResult, arbErr := c.arbiter.Arbitrate(ctx, batch, set)
	if arbErr != nil {
		return set, ArbitrationResult{}, errors.New("streamgate: evaluate arbiter: " + arbErr.Error())
	}

	return set, arbResult, nil
}

// Submit enqueues an epoch evaluation job. The caller blocks until the job is
// accepted into the bounded queue, until the caller's context is cancelled,
// or until the coordinator is closed. The job is processed by the single
// worker in FIFO order.
func (c *GateCoordinator) Submit(ctx context.Context, batch EvidenceBatch, filters []EpochFilter) (ArbitrationResult, error) {
	if err := batch.Validate(); err != nil {
		return ArbitrationResult{}, errors.New("streamgate: submit batch: " + err.Error())
	}

	reply := make(chan epochResult, 1)
	job := epochJob{
		ctx:     ctx,
		batch:   batch,
		filters: filters,
		reply:   reply,
	}

	select {
	case c.jobs <- job:
		// Job accepted into the bounded queue.
	case <-ctx.Done():
		return ArbitrationResult{}, ctx.Err()
	case <-c.closedCh:
		return ArbitrationResult{}, errors.New("streamgate: submit rejected: coordinator closed")
	}

	select {
	case res := <-reply:
		return res.result, res.err
	case <-ctx.Done():
		return ArbitrationResult{}, ctx.Err()
	case <-c.closedCh:
		return ArbitrationResult{}, errors.New("streamgate: submit rejected: coordinator closed")
	}
}

// Close terminates the coordinator and releases any blocked submitters.
// It is safe to call Close multiple times; only the first call performs the
// actual shutdown.
func (c *GateCoordinator) Close() error {
	var firstErr error
	c.closeOnce.Do(func() {
		c.closedOnce.Do(func() { close(c.closedCh) })
		close(c.jobs)
	})
	c.wg.Wait()
	return firstErr
}