Policy Engine

The Policy Engine is the authoritative decision-maker for retry, recovery, scaling, and operational automation.

Interface

public interface PolicyEngine {
    List<PolicyDecision> evaluate(PipelineState state, List<MetricsEnvelope> metrics);
    void registerPolicy(PolicyDefinition policy);
    RetryDecision handleFailure(StageFailure failure);
    ScalingAction handleScalingTrigger(ScalingTrigger trigger);
    QualityAction handleQualityBreach(QualityBreach breach);
    CatchUpPlan scheduleCatchUp(String pipeline, String fromDate, String toDate);
}

Policy Actions

Action	Description
RETRY	Retry with configured backoff strategy
SCALE	Scale resources up or down
ALERT	Send notifications to configured channels
QUARANTINE	Route failing data to quarantine destination
HALT	Stop pipeline execution
RESTART_FROM_CHECKPOINT	Resume from last successful checkpoint

Registering Policies

PolicyDefinition policy = new PolicyDefinition(
    "high-failure-rate-alert",
    "metrics.error_rate > threshold",
    "error_rate > 0.05",
    new PolicyAction(ActionType.ALERT, Map.of("channels", List.of("slack", "pagerduty"))),
    Duration.ofMinutes(5)  // evaluation interval
);

policyEngine.registerPolicy(policy);

Handling Failures

StageFailure failure = new StageFailure(
    "daily-ingest", "transform-stage",
    "TIMEOUT", "Stage exceeded PT15M timeout",
    true,  // retryable
    2      // attempt count so far
);

RetryDecision decision = policyEngine.handleFailure(failure);
// decision.shouldRetry() == true
// decision.delay() == Duration.ofSeconds(30)
// decision.maxAttempts() == 5

Handling Quality Breaches

QualityBreach breach = new QualityBreach(
    "daily-ingest", "customer-orders",
    0.85,   // observed quality score
    0.95,   // configured threshold
    List.of("not-null-check", "range-check")
);

QualityAction action = policyEngine.handleQualityBreach(breach);
// action.type() == "quarantine"
// action.destination() == "quarantine/customer-orders"

Catch-Up Scheduling

When pipelines miss business dates, schedule catch-up:

CatchUpPlan plan = policyEngine.scheduleCatchUp(
    "daily-ingest", "2024-01-10", "2024-01-15"
);
// plan.businessDates() == ["2024-01-10", "2024-01-11", ..., "2024-01-15"]
// plan.parallelism() == 3 (run up to 3 dates in parallel)

Policy Evaluation Flow

sequenceDiagram
    participant Echo
    participant PE as Policy Engine
    participant Provider

    Echo->>PE: State change + metrics
    PE->>PE: Evaluate registered policies
    PE->>PE: Match conditions
    alt Retry needed
        PE->>Provider: Retry with backoff
    else Scale needed
        PE->>Provider: Scale resources
    else Alert needed
        PE->>Echo: Publish alert event
    else Halt needed
        PE->>Provider: Stop execution
    end