Documentation
¶
Overview ¶
Package backpressure provides quality validation implementations
Package backpressure provides adaptive concurrency control for Drover ¶
Package backpressure provides quality validation and backpressure mechanisms
Index ¶
- type Check
- type CheckResult
- type CheckType
- type Config
- type Controller
- func (c *Controller) CanSpawn() bool
- func (c *Controller) GetBackoffDeadline() time.Time
- func (c *Controller) GetCurrentConcurrency() int
- func (c *Controller) GetCurrentInFlight() int
- func (c *Controller) GetStats() Stats
- func (c *Controller) IsInBackoff() bool
- func (c *Controller) OnWorkerSignal(signal WorkerSignal)
- func (c *Controller) Reset()
- func (c *Controller) WorkerFinished()
- func (c *Controller) WorkerStarted()
- type ControllerConfig
- type Manager
- func (m *Manager) CreateRule(rule *Rule) error
- func (m *Manager) EnableCheck(id string, enabled bool) error
- func (m *Manager) EnableRule(id string, enabled bool) error
- func (m *Manager) GetCheck(id string) (*Check, bool)
- func (m *Manager) GetRule(id string) (*Rule, bool)
- func (m *Manager) ListChecks() []*Check
- func (m *Manager) ListRules() []*Rule
- func (m *Manager) LoadFromFile(path string) error
- func (m *Manager) RegisterCheck(check *Check) error
- func (m *Manager) SaveToFile(path string) error
- func (m *Manager) SetLogger(logger *log.Logger)
- func (m *Manager) Validate(ctx context.Context, ruleID string, input *ValidateInput) (*RuleResult, error)
- type Rule
- type RuleResult
- type Stats
- type ValidateInput
- type WorkerSignal
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Check ¶
type Check struct {
ID string `json:"id"`
Name string `json:"name"`
Type CheckType `json:"type"`
Description string `json:"description"`
Enabled bool `json:"enabled"`
Config map[string]interface{} `json:"config"`
CreatedAt int64 `json:"created_at"`
UpdatedAt int64 `json:"updated_at"`
}
Check defines a validation check
type CheckResult ¶
type CheckResult struct {
CheckID string `json:"check_id"`
Passed bool `json:"passed"`
Message string `json:"message"`
Duration time.Duration `json:"duration"`
Timestamp int64 `json:"timestamp"`
Metadata interface{} `json:"metadata,omitempty"`
}
CheckResult is the result of a validation check
type Controller ¶
type Controller struct {
// contains filtered or unexported fields
}
Controller manages adaptive concurrency based on downstream health
func NewController ¶
func NewController(cfg ControllerConfig) *Controller
NewController creates a new backpressure controller
func (*Controller) CanSpawn ¶
func (c *Controller) CanSpawn() bool
CanSpawn checks if a new worker can be spawned based on backpressure state
func (*Controller) GetBackoffDeadline ¶
func (c *Controller) GetBackoffDeadline() time.Time
GetBackoffDeadline returns when the backoff period ends
func (*Controller) GetCurrentConcurrency ¶
func (c *Controller) GetCurrentConcurrency() int
GetCurrentConcurrency returns the current max concurrency
func (*Controller) GetCurrentInFlight ¶
func (c *Controller) GetCurrentInFlight() int
GetCurrentInFlight returns the current number of in-flight workers
func (*Controller) GetStats ¶
func (c *Controller) GetStats() Stats
GetStats returns current statistics
func (*Controller) IsInBackoff ¶
func (c *Controller) IsInBackoff() bool
IsInBackoff returns true if currently in backoff period
func (*Controller) OnWorkerSignal ¶
func (c *Controller) OnWorkerSignal(signal WorkerSignal)
OnWorkerSignal processes a worker signal and adjusts backpressure state
func (*Controller) Reset ¶
func (c *Controller) Reset()
Reset resets the controller to initial state
func (*Controller) WorkerFinished ¶
func (c *Controller) WorkerFinished()
WorkerFinished decrements the in-flight counter
func (*Controller) WorkerStarted ¶
func (c *Controller) WorkerStarted()
WorkerStarted increments the in-flight counter
type ControllerConfig ¶
type ControllerConfig struct {
InitialConcurrency int // Starting concurrency level
MinConcurrency int // Minimum concurrency (never go below)
MaxConcurrency int // Maximum concurrency (never exceed)
RateLimitBackoff time.Duration // Initial backoff on rate limit
MaxBackoff time.Duration // Maximum backoff duration
SlowThreshold time.Duration // Response time considered slow
SlowCountThreshold int // Consecutive slow responses before reducing
// Memory-aware settings (drover-mem-6)
MemoryAwareEnabled bool // Enable memory-aware spawning
MemoryThresholdMB int64 // Minimum available MB before throttling
MemoryCriticalMB int64 // Critical memory threshold - stop spawning
WorkerRSSLimitMB int64 // Per-worker RSS limit in MB
}
ControllerConfig holds backpressure controller configuration
func DefaultControllerConfig ¶
func DefaultControllerConfig() ControllerConfig
DefaultControllerConfig returns default backpressure controller configuration
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager manages backpressure checks and rules
func NewManager ¶
NewManager creates a new backpressure manager
func (*Manager) CreateRule ¶
CreateRule creates a new rule
func (*Manager) EnableCheck ¶
EnableCheck enables a check
func (*Manager) EnableRule ¶
EnableRule enables or disables a rule
func (*Manager) LoadFromFile ¶
LoadFromFile loads checks and rules from a JSON file
func (*Manager) RegisterCheck ¶
RegisterCheck registers a new validation check
func (*Manager) SaveToFile ¶
SaveToFile saves checks and rules to a JSON file
func (*Manager) Validate ¶
func (m *Manager) Validate(ctx context.Context, ruleID string, input *ValidateInput) (*RuleResult, error)
Validate runs all enabled checks in a rule and returns the result
type Rule ¶
type Rule struct {
ID string `json:"id"`
Name string `json:"name"`
Description string `json:"description"`
Enabled bool `json:"enabled"`
CheckIDs []string `json:"check_ids"` // Checks to run
Mode string `json:"mode"` // "all" (all must pass) or "any" (one must pass)
CreatedAt int64 `json:"created_at"`
UpdatedAt int64 `json:"updated_at"`
}
Rule defines a backpressure rule
type RuleResult ¶
type RuleResult struct {
RuleID string `json:"rule_id"`
Passed bool `json:"passed"`
CheckResults []*CheckResult `json:"check_results"`
Timestamp int64 `json:"timestamp"`
}
RuleResult is the result of a rule evaluation
type Stats ¶
type Stats struct {
MaxInFlight int // Current max concurrency
CurrentInFlight int // Currently active workers
BackoffUntil time.Time // When backoff ends
InBackoff bool // Currently in backoff
ConsecutiveSlow int // Count of slow responses
}
GetStats returns current controller statistics
type ValidateInput ¶
type ValidateInput struct {
TaskID string `json:"task_id"`
Title string `json:"title"`
Description string `json:"description"`
Output string `json:"output"`
ProjectDir string `json:"project_dir"`
Metadata map[string]interface{} `json:"metadata"`
}
ValidateInput is input for validation
type WorkerSignal ¶
type WorkerSignal string
WorkerSignal represents downstream health signals from workers
const ( SignalOK WorkerSignal = "ok" // Normal execution SignalRateLimited WorkerSignal = "rate_limited" // Rate limit detected SignalSlowResponse WorkerSignal = "slow_response" // Slow response SignalAPIError WorkerSignal = "api_error" // Transient API error )
Source Files