691 lines
22 KiB
Go
691 lines
22 KiB
Go
package streamgate
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
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()
|
|
}
|
|
|
|
// attemptCancellerAdapter wraps RequestRuntime to satisfy StreamReleaser's AttemptCanceller seam.
|
|
type attemptCancellerAdapter struct {
|
|
r *RequestRuntime
|
|
}
|
|
|
|
func (a *attemptCancellerAdapter) CancelAttempt(ctx context.Context, attemptID string) error {
|
|
if a.r != nil {
|
|
a.r.mu.Lock()
|
|
ctrl := a.r.currentBinding.Controller()
|
|
a.r.mu.Unlock()
|
|
if ctrl != nil {
|
|
return ctrl.AbortAttempt(ctx)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// RequestRuntime drives the request-local lifecycle loop, managing attempt bindings,
|
|
// evidence collection, parallel filter evaluation through GateCoordinator, decision arbitration,
|
|
// stream release, and fault recovery.
|
|
type RequestRuntime struct {
|
|
snapshot RequestRuntimeSnapshot
|
|
modelGroup string
|
|
currentBinding AttemptBinding
|
|
tail *EvidenceTail
|
|
boundary *CommitBoundary
|
|
releaser *StreamReleaser
|
|
gate *GateCoordinator
|
|
arbiter *DecisionArbiter
|
|
recoveryCoord *RecoveryCoordinator
|
|
recoveryPolicy RecoveryPolicySnapshot
|
|
recoveryUsage RecoveryUsageSnapshot
|
|
resolvedFilters []ResolvedFilter
|
|
stagedStart *ResponseStart
|
|
terminalCommitted bool
|
|
running bool
|
|
planIDCounter int
|
|
mu sync.Mutex
|
|
}
|
|
|
|
// NewRequestRuntime constructs a new RequestRuntime instance.
|
|
func NewRequestRuntime(snapshot RequestRuntimeSnapshot, modelGroup string, initial AttemptBinding) (*RequestRuntime, error) {
|
|
if err := snapshot.Validate(); err != nil {
|
|
return nil, fmt.Errorf("streamgate: new request runtime snapshot invalid: %w", err)
|
|
}
|
|
if modelGroup == "" {
|
|
return nil, errors.New("streamgate: new request runtime model group is required")
|
|
}
|
|
if err := initial.Validate(); err != nil {
|
|
return nil, fmt.Errorf("streamgate: new request runtime initial binding invalid: %w", err)
|
|
}
|
|
|
|
boundary, err := NewCommitBoundary(snapshot.Sink())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
arbiter := NewDecisionArbiter()
|
|
gate, err := NewGateCoordinator(context.Background(), snapshot.Options().GateOptions(), arbiter)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
recPolicy, err := NewRecoveryPolicySnapshot(snapshot.Options().MaxRecoveryAttemptsTotal(), nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
recUsage, err := NewRecoveryUsageSnapshot(0, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
preparers := make(map[string]RecoveryPlanPreparer)
|
|
if snapshot.Preparer() != nil {
|
|
preparers["default"] = snapshot.Preparer()
|
|
}
|
|
|
|
recOpts := RecoveryCoordinatorOptions{
|
|
Policy: recPolicy,
|
|
Usage: recUsage,
|
|
RequestSnapshot: snapshot.RequestSnapshotRef(),
|
|
CurrentBinding: initial,
|
|
Rebuilder: snapshot.Rebuilder(),
|
|
Dispatcher: snapshot.Dispatcher(),
|
|
Preparers: preparers,
|
|
PreparationTimeout: 5 * time.Second,
|
|
}
|
|
|
|
recCoord, err := NewRecoveryCoordinator(recOpts)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
rt := &RequestRuntime{
|
|
snapshot: snapshot,
|
|
modelGroup: modelGroup,
|
|
boundary: boundary,
|
|
gate: gate,
|
|
arbiter: arbiter,
|
|
recoveryCoord: recCoord,
|
|
recoveryPolicy: recPolicy,
|
|
recoveryUsage: recUsage,
|
|
}
|
|
|
|
if err := rt.installAttempt(initial, RecoveryResumeModeReplaceAttempt); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return rt, nil
|
|
}
|
|
|
|
func (r *RequestRuntime) installAttempt(binding AttemptBinding, resumeMode RecoveryResumeMode) error {
|
|
if err := binding.Validate(); err != nil {
|
|
return err
|
|
}
|
|
|
|
oldAttemptID := r.currentBinding.AttemptID()
|
|
r.currentBinding = binding
|
|
|
|
target, err := NewAttemptTarget(r.modelGroup, binding.Model(), binding.Provider(), binding.ExecutionPath(), nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
rfc, err := NewRequestFilterContext(
|
|
r.snapshot.ConfigGeneration(),
|
|
binding.AttemptID(),
|
|
r.snapshot.Environment(),
|
|
r.snapshot.Endpoint(),
|
|
r.snapshot.Family(),
|
|
"",
|
|
r.boundary.State(),
|
|
false,
|
|
false,
|
|
r.snapshot.RequestID(),
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
rfSnap, err := r.snapshot.RegistrySnapshot().BeginRequest(rfc)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
resolved, err := rfSnap.ResolveAttempt(target)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
r.resolvedFilters = resolved
|
|
|
|
plan, err := NewEvidencePlanFromResolvedFilters(resolved)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if r.tail == nil {
|
|
r.tail, err = NewEvidenceTail(plan)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
r.releaser, err = NewStreamReleaser(r.tail, r.boundary, &attemptCancellerAdapter{r: r})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := r.boundary.BeginAttempt(binding.AttemptID()); err != nil {
|
|
return err
|
|
}
|
|
} else if resumeMode == RecoveryResumeModeReplaceAttempt {
|
|
if oldAttemptID != "" {
|
|
_ = r.releaser.ReplaceUncommittedAttempt(context.Background(), oldAttemptID, binding.AttemptID())
|
|
} else {
|
|
_ = r.boundary.BeginAttempt(binding.AttemptID())
|
|
r.tail.ResetForReplace()
|
|
}
|
|
r.tail, err = NewEvidenceTail(plan)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
r.releaser, err = NewStreamReleaser(r.tail, r.boundary, &attemptCancellerAdapter{r: r})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
} else if resumeMode == RecoveryResumeModeContinueStream {
|
|
r.tail.PrepareContinuation()
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Run executes the normalized event owner loop for the request lifetime.
|
|
func (r *RequestRuntime) Run(ctx context.Context) error {
|
|
r.mu.Lock()
|
|
if r.running || r.terminalCommitted {
|
|
r.mu.Unlock()
|
|
return errors.New("streamgate: runtime already running or terminal")
|
|
}
|
|
r.running = true
|
|
r.mu.Unlock()
|
|
|
|
defer func() {
|
|
if r.gate != nil {
|
|
_ = r.gate.Close()
|
|
}
|
|
}()
|
|
|
|
for {
|
|
r.mu.Lock()
|
|
if r.terminalCommitted {
|
|
r.mu.Unlock()
|
|
return nil
|
|
}
|
|
binding := r.currentBinding
|
|
r.mu.Unlock()
|
|
|
|
if ctx.Err() != nil {
|
|
return ctx.Err()
|
|
}
|
|
|
|
ev, err := binding.EventSource().NextEvent(ctx)
|
|
if err != nil {
|
|
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || ctx.Err() != nil {
|
|
return ctx.Err()
|
|
}
|
|
r.mu.Lock()
|
|
if !r.terminalCommitted {
|
|
termRes, _ := NewSuccessTerminalResult("default", time.Now())
|
|
_ = r.boundary.CommitTerminal(ctx, binding.AttemptID(), termRes)
|
|
r.terminalCommitted = true
|
|
}
|
|
r.mu.Unlock()
|
|
return nil
|
|
}
|
|
|
|
if ev.Kind() == EventKindResponseStart {
|
|
rs, err := ev.AsResponseStart()
|
|
if err == nil {
|
|
r.stagedStart = &rs
|
|
_ = r.boundary.StageResponseStart(binding.AttemptID(), rs)
|
|
}
|
|
continue
|
|
}
|
|
|
|
epoch, signal, err := r.tail.Append(ev)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if signal == EvidenceTailSignalBufferOverflow {
|
|
desc, _ := NewExternalDescriptor("error", "buffer_overflow", "buffer overflow occurred", "")
|
|
cause, _ := NewFailureCause("evidence_tail", "buffer_overflow", "", "", "")
|
|
causes, _ := NewFailureCauseChain([]FailureCause{cause})
|
|
termRes, _ := NewErrorTerminalResult(ev.Channel(), desc, causes, time.Now())
|
|
_, _ = r.releaser.FailPending(ctx, binding.AttemptID(), termRes)
|
|
r.mu.Lock()
|
|
r.terminalCommitted = true
|
|
r.mu.Unlock()
|
|
return nil
|
|
}
|
|
|
|
if signal == EvidenceTailSignalThreshold || signal == EvidenceTailSignalTrigger || signal == EvidenceTailSignalReady {
|
|
var epochFilters []EpochFilter
|
|
for _, rf := range r.resolvedFilters {
|
|
app, ok := epoch.ApplicabilityFor(rf.FilterID())
|
|
if ok {
|
|
ef, err := rf.BindEpoch(epoch.ID(), app)
|
|
if err == nil {
|
|
epochFilters = append(epochFilters, ef)
|
|
}
|
|
}
|
|
}
|
|
|
|
terminalFlag := (ev.Kind() == EventKindTerminal || ev.Kind() == EventKindProviderError)
|
|
stagedStartPtr := r.stagedStart
|
|
if r.boundary.State() != CommitStateTransportUncommitted {
|
|
stagedStartPtr = nil
|
|
}
|
|
|
|
batch, err := NewEvidenceBatch(
|
|
[]NormalizedEvent{ev},
|
|
nil,
|
|
nil,
|
|
stagedStartPtr,
|
|
terminalFlag,
|
|
r.boundary.State(),
|
|
time.Now(),
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
arbResult, err := r.gate.Submit(ctx, batch, epochFilters)
|
|
if err != nil {
|
|
if ctx.Err() != nil {
|
|
return ctx.Err()
|
|
}
|
|
return err
|
|
}
|
|
|
|
switch arbResult.Action() {
|
|
case ArbitrationActionRelease:
|
|
if terminalFlag {
|
|
termRes, _ := NewSuccessTerminalResult(ev.Channel(), time.Now())
|
|
_, err := r.releaser.ReleaseTerminalEpoch(ctx, binding.AttemptID(), epoch.ID(), termRes)
|
|
if err != nil {
|
|
_ = r.boundary.CommitTerminal(ctx, binding.AttemptID(), termRes)
|
|
}
|
|
r.mu.Lock()
|
|
r.terminalCommitted = true
|
|
r.mu.Unlock()
|
|
return nil
|
|
}
|
|
_, err := r.releaser.ReleaseEpoch(ctx, binding.AttemptID(), epoch.ID())
|
|
if err != nil {
|
|
relEv, relErr := normalizedToReleaseEvent(ev)
|
|
if relErr == nil && relEv.Kind() != "" {
|
|
_, _ = r.boundary.ReleaseSafe(ctx, binding.AttemptID(), []ReleaseEvent{relEv})
|
|
}
|
|
}
|
|
|
|
case ArbitrationActionHold:
|
|
// Hold downstream; events stay buffered in tail
|
|
|
|
case ArbitrationActionReplacement:
|
|
if terminalFlag {
|
|
termRes, _ := NewSuccessTerminalResult(ev.Channel(), time.Now())
|
|
_, _ = r.releaser.ReleaseTerminalEpoch(ctx, binding.AttemptID(), epoch.ID(), termRes)
|
|
r.mu.Lock()
|
|
r.terminalCommitted = true
|
|
r.mu.Unlock()
|
|
return nil
|
|
}
|
|
_, _ = r.releaser.ReleaseEpoch(ctx, binding.AttemptID(), epoch.ID())
|
|
|
|
case ArbitrationActionRecover:
|
|
r.planIDCounter++
|
|
planID := fmt.Sprintf("plan-%s-%d", r.snapshot.RequestID(), r.planIDCounter)
|
|
idempotencyKey := fmt.Sprintf("idemp-%s-%d", r.snapshot.RequestID(), r.planIDCounter)
|
|
preparerID := ""
|
|
if r.snapshot.Preparer() != nil {
|
|
preparerID = "default"
|
|
}
|
|
|
|
cycleInput := RecoveryCycleInput{
|
|
Arbitration: arbResult,
|
|
PlanID: planID,
|
|
IdempotencyKey: idempotencyKey,
|
|
ConsumerID: arbResult.FilterID(),
|
|
CommitState: r.boundary.State(),
|
|
CallerCanceled: ctx.Err() != nil,
|
|
PreparerID: preparerID,
|
|
PreparationSnapshot: nil,
|
|
}
|
|
|
|
cycleRes, err := r.recoveryCoord.Execute(ctx, cycleInput)
|
|
newBinding, hasBinding := cycleRes.Binding()
|
|
plan, _ := cycleRes.Plan()
|
|
if err != nil || !hasBinding {
|
|
desc, _ := NewExternalDescriptor("error", "recovery_failed", "recovery failed", "")
|
|
causes := cycleRes.FailureCauses()
|
|
if causes.Len() == 0 {
|
|
c, _ := NewFailureCause("recovery", "recovery_failed", "", "", "")
|
|
causes, _ = NewFailureCauseChain([]FailureCause{c})
|
|
}
|
|
termRes, _ := NewErrorTerminalResult(ev.Channel(), desc, causes, time.Now())
|
|
_, _ = r.releaser.FailPending(ctx, binding.AttemptID(), termRes)
|
|
r.mu.Lock()
|
|
r.terminalCommitted = true
|
|
r.mu.Unlock()
|
|
return nil
|
|
}
|
|
|
|
if err := r.installAttempt(newBinding, plan.ResumeMode()); err != nil {
|
|
return err
|
|
}
|
|
|
|
case ArbitrationActionTerminal:
|
|
desc, _ := NewExternalDescriptor("error", "fatal_violation", "fatal filter violation", "")
|
|
cause, _ := NewFailureCause("arbiter", "fatal_violation", arbResult.FilterID(), arbResult.RuleID(), "")
|
|
causes, _ := NewFailureCauseChain([]FailureCause{cause})
|
|
termRes, _ := NewErrorTerminalResult(ev.Channel(), desc, causes, time.Now())
|
|
_, _ = r.releaser.FailPending(ctx, binding.AttemptID(), termRes)
|
|
r.mu.Lock()
|
|
r.terminalCommitted = true
|
|
r.mu.Unlock()
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
}
|