451 lines
14 KiB
Go
451 lines
14 KiB
Go
package streamgate
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
)
|
|
|
|
// ArbitrationAction identifies the single deterministic action returned by
|
|
// the decision arbiter for an epoch.
|
|
type ArbitrationAction string
|
|
|
|
const (
|
|
// ArbitrationActionRelease indicates the batch is eligible for release.
|
|
ArbitrationActionRelease ArbitrationAction = "release"
|
|
|
|
// ArbitrationActionHold indicates the batch must be buffered downstream.
|
|
ArbitrationActionHold ArbitrationAction = "hold"
|
|
|
|
// ArbitrationActionTerminal indicates the stream must be terminated.
|
|
ArbitrationActionTerminal ArbitrationAction = "terminal"
|
|
|
|
// ArbitrationActionRecover indicates a recovery workflow must be triggered.
|
|
ArbitrationActionRecover ArbitrationAction = "recover"
|
|
|
|
// ArbitrationActionReplacement indicates replacement content must be substituted.
|
|
ArbitrationActionReplacement ArbitrationAction = "replacement"
|
|
)
|
|
|
|
var knownArbitrationActions = map[ArbitrationAction]struct{}{
|
|
ArbitrationActionRelease: {},
|
|
ArbitrationActionHold: {},
|
|
ArbitrationActionTerminal: {},
|
|
ArbitrationActionRecover: {},
|
|
ArbitrationActionReplacement: {},
|
|
}
|
|
|
|
// Validate returns nil when the ArbitrationAction is a known value.
|
|
func (a ArbitrationAction) Validate() error {
|
|
if _, ok := knownArbitrationActions[a]; ok {
|
|
return nil
|
|
}
|
|
return errors.New("streamgate: unknown arbitration action: " + string(a))
|
|
}
|
|
|
|
// ArbitrationResult is the immutable, closed output produced by DecisionArbiter.
|
|
type ArbitrationResult struct {
|
|
action ArbitrationAction
|
|
baseDisposition BaseEventDisposition
|
|
filterID string
|
|
ruleID string
|
|
intent *RecoveryIntent
|
|
replacement *ReplacementProposal
|
|
}
|
|
|
|
// NewArbitrationResult constructs a validated, immutable ArbitrationResult.
|
|
func NewArbitrationResult(
|
|
action ArbitrationAction,
|
|
baseDisp BaseEventDisposition,
|
|
filterID string,
|
|
ruleID string,
|
|
intent *RecoveryIntent,
|
|
replacement *ReplacementProposal,
|
|
) (ArbitrationResult, error) {
|
|
if err := action.Validate(); err != nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: arbitration result action: " + err.Error())
|
|
}
|
|
if err := baseDisp.Validate(); err != nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: arbitration result base disposition: " + err.Error())
|
|
}
|
|
if filterID != "" {
|
|
if err := validateStableTokenRequired("filterID", filterID); err != nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: arbitration result filter id: " + err.Error())
|
|
}
|
|
}
|
|
if ruleID != "" {
|
|
if filterID == "" {
|
|
return ArbitrationResult{}, errors.New("streamgate: arbitration result rule id requires filter id")
|
|
}
|
|
if err := validateStableTokenRequired("ruleID", ruleID); err != nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: arbitration result rule id: " + err.Error())
|
|
}
|
|
}
|
|
|
|
var intentCopy *RecoveryIntent
|
|
if intent != nil {
|
|
if err := intent.Validate(); err != nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: arbitration result recovery intent: " + err.Error())
|
|
}
|
|
cp := *intent
|
|
intentCopy = &cp
|
|
}
|
|
|
|
var replacementCopy *ReplacementProposal
|
|
if replacement != nil {
|
|
if err := replacement.Validate(); err != nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: arbitration result replacement proposal: " + err.Error())
|
|
}
|
|
cp := cloneReplacementProposal(replacement)
|
|
replacementCopy = cp
|
|
}
|
|
|
|
switch action {
|
|
case ArbitrationActionRelease, ArbitrationActionHold:
|
|
if filterID != "" {
|
|
return ArbitrationResult{}, errors.New("streamgate: release and hold actions must not carry filter id")
|
|
}
|
|
if ruleID != "" {
|
|
return ArbitrationResult{}, errors.New("streamgate: release and hold actions must not carry rule id")
|
|
}
|
|
if intentCopy != nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: release and hold actions must not carry recovery intent")
|
|
}
|
|
if replacementCopy != nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: release and hold actions must not carry replacement proposal")
|
|
}
|
|
case ArbitrationActionTerminal:
|
|
if intentCopy != nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: terminal action must not carry recovery intent")
|
|
}
|
|
if replacementCopy != nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: terminal action must not carry replacement proposal")
|
|
}
|
|
case ArbitrationActionRecover:
|
|
if filterID == "" {
|
|
return ArbitrationResult{}, errors.New("streamgate: recover action requires filter id")
|
|
}
|
|
if ruleID == "" {
|
|
return ArbitrationResult{}, errors.New("streamgate: recover action requires rule id")
|
|
}
|
|
if intentCopy == nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: recover action requires recovery intent")
|
|
}
|
|
if replacementCopy != nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: recover action must not carry replacement proposal")
|
|
}
|
|
case ArbitrationActionReplacement:
|
|
if filterID == "" {
|
|
return ArbitrationResult{}, errors.New("streamgate: replacement action requires filter id")
|
|
}
|
|
if ruleID == "" {
|
|
return ArbitrationResult{}, errors.New("streamgate: replacement action requires rule id")
|
|
}
|
|
if replacementCopy == nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: replacement action requires replacement proposal")
|
|
}
|
|
if intentCopy != nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: replacement action must not carry recovery intent")
|
|
}
|
|
}
|
|
|
|
return ArbitrationResult{
|
|
action: action,
|
|
baseDisposition: baseDisp,
|
|
filterID: filterID,
|
|
ruleID: ruleID,
|
|
intent: intentCopy,
|
|
replacement: replacementCopy,
|
|
}, nil
|
|
}
|
|
|
|
// Validate returns nil when the ArbitrationResult is in a consistent state.
|
|
func (r ArbitrationResult) Validate() error {
|
|
if err := r.action.Validate(); err != nil {
|
|
return err
|
|
}
|
|
if err := r.baseDisposition.Validate(); err != nil {
|
|
return err
|
|
}
|
|
if r.filterID != "" {
|
|
if err := validateStableTokenRequired("filterID", r.filterID); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if r.ruleID != "" {
|
|
if r.filterID == "" {
|
|
return errors.New("streamgate: arbitration result rule id requires filter id")
|
|
}
|
|
if err := validateStableTokenRequired("ruleID", r.ruleID); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
switch r.action {
|
|
case ArbitrationActionRelease, ArbitrationActionHold:
|
|
if r.filterID != "" || r.ruleID != "" || r.intent != nil || r.replacement != nil {
|
|
return errors.New("streamgate: release and hold actions must not carry filter, rule, intent, or replacement")
|
|
}
|
|
case ArbitrationActionTerminal:
|
|
if r.intent != nil || r.replacement != nil {
|
|
return errors.New("streamgate: terminal action must not carry intent or replacement")
|
|
}
|
|
case ArbitrationActionRecover:
|
|
if r.filterID == "" || r.ruleID == "" || r.intent == nil || r.replacement != nil {
|
|
return errors.New("streamgate: recover action requires filter id, rule id, and intent, and no replacement")
|
|
}
|
|
if err := r.intent.Validate(); err != nil {
|
|
return err
|
|
}
|
|
case ArbitrationActionReplacement:
|
|
if r.filterID == "" || r.ruleID == "" || r.replacement == nil || r.intent != nil {
|
|
return errors.New("streamgate: replacement action requires filter id, rule id, and replacement proposal, and no intent")
|
|
}
|
|
if err := r.replacement.Validate(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Action returns the derived arbitration action.
|
|
func (r ArbitrationResult) Action() ArbitrationAction { return r.action }
|
|
|
|
// BaseDisposition returns the original base event disposition.
|
|
func (r ArbitrationResult) BaseDisposition() BaseEventDisposition { return r.baseDisposition }
|
|
|
|
// FilterID returns the selected filter id, or empty.
|
|
func (r ArbitrationResult) FilterID() string { return r.filterID }
|
|
|
|
// RuleID returns the selected rule id, or empty.
|
|
func (r ArbitrationResult) RuleID() string { return r.ruleID }
|
|
|
|
// RecoveryIntent returns a defensive copy of the recovery intent, or nil.
|
|
func (r ArbitrationResult) RecoveryIntent() *RecoveryIntent {
|
|
if r.intent == nil {
|
|
return nil
|
|
}
|
|
cp := *r.intent
|
|
return &cp
|
|
}
|
|
|
|
// ReplacementProposal returns a defensive copy of the replacement proposal, or nil.
|
|
func (r ArbitrationResult) ReplacementProposal() *ReplacementProposal {
|
|
return cloneReplacementProposal(r.replacement)
|
|
}
|
|
|
|
// DecisionArbiter implements EpochArbiter using the closed S13 decision matrix.
|
|
type DecisionArbiter struct{}
|
|
|
|
// NewDecisionArbiter creates a new DecisionArbiter instance.
|
|
func NewDecisionArbiter() *DecisionArbiter {
|
|
return &DecisionArbiter{}
|
|
}
|
|
|
|
// Arbitrate synthesizes the evidence batch base disposition and all active
|
|
// filter outcomes into a single deterministic ArbitrationResult.
|
|
func (a *DecisionArbiter) Arbitrate(
|
|
ctx context.Context,
|
|
batch EvidenceBatch,
|
|
set EvaluationSet,
|
|
) (ArbitrationResult, error) {
|
|
return Arbitrate(ctx, batch, set)
|
|
}
|
|
|
|
// Arbitrate synthesizes the evidence batch base disposition and all active
|
|
// filter outcomes into a single deterministic ArbitrationResult.
|
|
func Arbitrate(
|
|
ctx context.Context,
|
|
batch EvidenceBatch,
|
|
set EvaluationSet,
|
|
) (ArbitrationResult, error) {
|
|
if ctx.Err() != nil {
|
|
return ArbitrationResult{}, ctx.Err()
|
|
}
|
|
|
|
baseDisp, err := batch.BaseDisposition()
|
|
if err != nil {
|
|
return ArbitrationResult{}, fmt.Errorf("streamgate: arbitrate base disposition: %w", err)
|
|
}
|
|
|
|
if err := set.Validate(); err != nil {
|
|
return ArbitrationResult{}, fmt.Errorf("streamgate: arbitrate evaluation set: %w", err)
|
|
}
|
|
|
|
type candidate struct {
|
|
filterID string
|
|
ruleID string
|
|
priority int
|
|
kind string // "terminal", "recover", "replacement"
|
|
intent *RecoveryIntent
|
|
replacement *ReplacementProposal
|
|
}
|
|
|
|
var terminalFailures []candidate
|
|
var overrideCandidates []candidate
|
|
hasBlockingDeferred := false
|
|
|
|
for _, o := range set.All() {
|
|
if o.Enforcement() != FilterEnforcementBlocking {
|
|
// Observe-only outcomes are not blocking action candidates.
|
|
continue
|
|
}
|
|
|
|
switch o.Outcome().Kind() {
|
|
case FilterOutcomeKindEvaluated:
|
|
decision := o.Outcome().Decision()
|
|
if decision == nil {
|
|
return ArbitrationResult{}, errors.New("streamgate: arbitrate evaluated outcome missing decision")
|
|
}
|
|
switch decision.Kind() {
|
|
case FilterDecisionKindFatal:
|
|
terminalFailures = append(terminalFailures, candidate{
|
|
filterID: o.FilterID(),
|
|
ruleID: decision.RuleID(),
|
|
priority: o.Priority(),
|
|
kind: "terminal",
|
|
})
|
|
case FilterDecisionKindViolation:
|
|
intent := decision.RecoveryIntent()
|
|
if intent != nil {
|
|
if intent.Priority() != o.Priority() {
|
|
return ArbitrationResult{}, fmt.Errorf(
|
|
"streamgate: recovery intent priority %d mismatch with filter %s effective priority %d",
|
|
intent.Priority(), o.FilterID(), o.Priority(),
|
|
)
|
|
}
|
|
overrideCandidates = append(overrideCandidates, candidate{
|
|
filterID: o.FilterID(),
|
|
ruleID: decision.RuleID(),
|
|
priority: o.Priority(),
|
|
kind: "recover",
|
|
intent: intent,
|
|
})
|
|
} else {
|
|
terminalFailures = append(terminalFailures, candidate{
|
|
filterID: o.FilterID(),
|
|
ruleID: decision.RuleID(),
|
|
priority: o.Priority(),
|
|
kind: "terminal",
|
|
})
|
|
}
|
|
case FilterDecisionKindReplacement:
|
|
proposal := decision.ReplacementProposal()
|
|
overrideCandidates = append(overrideCandidates, candidate{
|
|
filterID: o.FilterID(),
|
|
ruleID: decision.RuleID(),
|
|
priority: o.Priority(),
|
|
kind: "replacement",
|
|
replacement: proposal,
|
|
})
|
|
case FilterDecisionKindPass, FilterDecisionKindObserve:
|
|
// Pass and observe do not override.
|
|
}
|
|
|
|
case FilterOutcomeKindEvaluationError:
|
|
terminalFailures = append(terminalFailures, candidate{
|
|
filterID: o.FilterID(),
|
|
ruleID: "",
|
|
priority: o.Priority(),
|
|
kind: "terminal",
|
|
})
|
|
|
|
case FilterOutcomeKindDeferredByRequirement:
|
|
if batch.TerminalFlag() {
|
|
terminalFailures = append(terminalFailures, candidate{
|
|
filterID: o.FilterID(),
|
|
ruleID: "",
|
|
priority: o.Priority(),
|
|
kind: "terminal",
|
|
})
|
|
} else {
|
|
hasBlockingDeferred = true
|
|
}
|
|
|
|
case FilterOutcomeKindNotApplicableForEpoch:
|
|
// Not applicable does not block release.
|
|
}
|
|
}
|
|
|
|
sortCandidates := func(c []candidate) {
|
|
sort.SliceStable(c, func(i, j int) bool {
|
|
if c[i].priority != c[j].priority {
|
|
return c[i].priority > c[j].priority // Descending priority
|
|
}
|
|
return c[i].filterID < c[j].filterID // Ascending filter id token
|
|
})
|
|
}
|
|
|
|
// 1. Any blocking terminal failure takes absolute precedence over recover/replacement.
|
|
if len(terminalFailures) > 0 {
|
|
sortCandidates(terminalFailures)
|
|
winner := terminalFailures[0]
|
|
return NewArbitrationResult(
|
|
ArbitrationActionTerminal,
|
|
baseDisp,
|
|
winner.filterID,
|
|
winner.ruleID,
|
|
nil,
|
|
nil,
|
|
)
|
|
}
|
|
|
|
// 2. Recover / Replacement overrides.
|
|
if len(overrideCandidates) > 0 {
|
|
sortCandidates(overrideCandidates)
|
|
winner := overrideCandidates[0]
|
|
if winner.kind == "recover" {
|
|
return NewArbitrationResult(
|
|
ArbitrationActionRecover,
|
|
baseDisp,
|
|
winner.filterID,
|
|
winner.ruleID,
|
|
winner.intent,
|
|
nil,
|
|
)
|
|
} else if winner.kind == "replacement" {
|
|
return NewArbitrationResult(
|
|
ArbitrationActionReplacement,
|
|
baseDisp,
|
|
winner.filterID,
|
|
winner.ruleID,
|
|
nil,
|
|
winner.replacement,
|
|
)
|
|
}
|
|
}
|
|
|
|
// 3. Blocking deferred requirement in a non-terminal batch yields hold.
|
|
if hasBlockingDeferred {
|
|
return NewArbitrationResult(
|
|
ArbitrationActionHold,
|
|
baseDisp,
|
|
"",
|
|
"",
|
|
nil,
|
|
nil,
|
|
)
|
|
}
|
|
|
|
// 4. Default base disposition action.
|
|
var defaultAction ArbitrationAction
|
|
switch baseDisp {
|
|
case BaseDispositionHold:
|
|
defaultAction = ArbitrationActionHold
|
|
case BaseDispositionReleaseCandidate:
|
|
defaultAction = ArbitrationActionRelease
|
|
case BaseDispositionTerminalSuccessCandidate, BaseDispositionTerminalErrorCandidate:
|
|
defaultAction = ArbitrationActionTerminal
|
|
default:
|
|
return ArbitrationResult{}, fmt.Errorf("streamgate: unhandled base disposition: %s", baseDisp)
|
|
}
|
|
|
|
return NewArbitrationResult(
|
|
defaultAction,
|
|
baseDisp,
|
|
"",
|
|
"",
|
|
nil,
|
|
nil,
|
|
)
|
|
}
|