iop/packages/go/streamgate/parallel_evaluation.go

package streamgate

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

// 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) 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
}

// 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
}

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

type epochResult struct {
	set EvaluationSet
	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{}),
	}
	c.wg.Add(1)
	go c.worker()
	return c, nil
}

// 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, err := c.Evaluate(job.ctx, job.batch, job.filters)
	select {
	case job.reply <- epochResult{set: set, 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, error) {
	if err := batch.Validate(); err != nil {
		return EvaluationSet{}, 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{}, ctx.Err()
	}

	// 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{}, ctx.Err()
		case <-c.hostCtx.Done():
			return EvaluationSet{}, c.hostCtx.Err()
		}
	}

	// 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(), oc, f.EnforcedAs())
			if err != nil {
				return EvaluationSet{}, 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(), readyOutcomes[i], w.filter.EnforcedAs())
		if err != nil {
			return EvaluationSet{}, 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{}, errors.New("streamgate: evaluate set: " + err.Error())
	}

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

	return set, 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) error {
	if err := batch.Validate(); err != nil {
		return 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 ctx.Err()
	case <-c.closedCh:
		return errors.New("streamgate: submit rejected: coordinator closed")
	}

	select {
	case res := <-reply:
		return res.err
	case <-ctx.Done():
		return ctx.Err()
	case <-c.closedCh:
		return 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
}