ETL Engine Scheduling

Overview

The ETL scheduling subsystem provides enterprise-grade pipeline orchestration:

• Trigger plugins - Cron, event-driven, webhook, and manual triggers
• Priority queues - Workload-class based queueing with concurrency limits
• Watermarks - Incremental sync with automatic window computation
• Backfills - Historical data reprocessing with windowing
• Circuit breakers - Automatic failure detection and isolation
• Dependency graphs - Schedule dependencies with cycle detection
• Rate limiting - Per-schedule execution throttling
• Governance loop - Dependency timeouts and fail-fast monitoring

Architecture

SchedulerHost (singleton)
├── ScheduleStorage          ← Persisted schedule definitions
├── WatermarkTracker         ← Incremental sync state
├── BackfillManager          ← Historical reprocessing
├── ScheduleAuditLog         ← Audit trail
├── PipelineRunQueue         ← Priority queue
├── ConcurrencyGate          ← Parallel execution limits
├── DependencyGraph          ← Schedule dependencies
└── PipelinePluginRegistry   ← Scheduler plugin discovery

DispatchLoopAsync
├── Dequeue run from queue
├── Check concurrency gate
├── Resolve dependencies
├── Execute via PipelineEngine
└── Update watermark / checkpoint

GovernanceLoopAsync (every 15s)
├── Check dependency timeouts
├── Detect stuck runs
├── Update circuit breaker state
└── Log governance events

SchedulerHost

Singleton service managing all scheduler plugins and pipeline execution:

public sealed class SchedulerHost : IAsyncDisposable
{
    public SchedulerHost(IDMEEditor editor);

    // Subsystems
    public ScheduleStorage Storage { get; }
    public WatermarkTracker Watermarks { get; }
    public BackfillManager Backfills { get; }
    public ScheduleAuditLog AuditLog { get; }

    // Events
    public event EventHandler<SchedulerRunEventArgs>? RunStarted;
    public event EventHandler<SchedulerRunEventArgs>? RunCompleted;
    public event EventHandler<SchedulerRunEventArgs>? RunFailed;
    public event EventHandler<SchedulerRunEventArgs>? CircuitBreakerChanged;

    // Lifecycle
    public Task StartAsync(CancellationToken token = default);
    public Task StopAsync();
    public Task ReloadAsync();

    // Manual trigger
    public Task<string> TriggerManualAsync(
        string pipelineOrWorkflowId,
        IReadOnlyDictionary<string, object>? overrideParams = null);

    // Backfill
    public Task<string> SubmitBackfillAsync(
        string scheduleId, string fromValue, string toValue,
        string reason = "", string requestedBy = "",
        int windowSizeSeconds = 86400);

    // Circuit breaker
    public Task ResetCircuitBreakerAsync(string scheduleId, string resetBy = "");

    // Monitoring
    public IReadOnlyList<(string RunId, string PipelineId, RunStatus Status)> GetActiveRuns();
}

Configuration

ScheduleDefinition

Persisted schedule configuration:

public class ScheduleDefinition
{
    public string Id { get; set; }
    public string Name { get; set; }
    public string PipelineId { get; set; }
    public bool IsWorkflow { get; set; }
    public bool IsEnabled { get; set; }
    public int Priority { get; set; } = 5;
    public string WorkloadClass { get; set; } = "standard";

    // Scheduler plugin
    public string SchedulerPluginId { get; set; }
    public Dictionary<string, object> SchedulerConfig { get; set; }

    // Watermark for incremental sync
    public WatermarkConfig Watermark { get; set; }

    // Dependencies
    public List<string> DependsOn { get; set; }
    public string DependencyCondition { get; set; } = "ALL_SUCCESS";
    public int DependencyMaxWaitSeconds { get; set; }

    // Circuit breaker
    public bool CircuitBreakerTripped { get; set; }
    public int ConsecutiveFailures { get; set; }
    public int CircuitBreakerThreshold { get; set; } = 5;
    public int CircuitBreakerCooldownSeconds { get; set; } = 300;

    // Rate limiting
    public int MaxRunsPerMinute { get; set; } = 0; // 0 = unlimited
}

PipelineRunQueue

Priority queue with workload-class isolation:

public sealed class PipelineRunQueue
{
    public int Count { get; }
    public int CountByClass(string workloadClass);

    // Concurrency limits per workload class
    public void SetClassConcurrencyLimit(string workloadClass, int maxConcurrent);

    // Queue operations
    public void Enqueue(QueuedRun run);
    public Task<QueuedRun> DequeueAsync(CancellationToken token);
    public void NotifyRunCompleted(string workloadClass);
}

QueuedRun

WatermarkTracker

Manages incremental sync state for each schedule:

public sealed class WatermarkTracker
{
    public WatermarkTracker(IDMEEditor editor);

    // Load/save watermark state
    public Task<WatermarkState?> LoadAsync(string scheduleId);
    public Task<IErrorsInfo> SaveAsync(string scheduleId, WatermarkState state);

    // Compute next window
    public WatermarkWindow ComputeWindow(
        WatermarkConfig config,
        WatermarkState? currentState);

    // Reset
    public Task<IErrorsInfo> ResetAsync(string scheduleId);
}

Watermark Types

BackfillManager

Historical data reprocessing with automatic windowing:

public sealed class BackfillManager
{
    public BackfillManager(IDMEEditor editor, WatermarkTracker watermarks);

    public Task<BackfillRequest> CreateRequestAsync(
        string scheduleId,
        string pipelineId,
        string fromValue,
        string toValue,
        string watermarkType = "datetime",
        string watermarkColumn = "",
        string reason = "",
        string requestedBy = "",
        int windowSizeSeconds = 86400);

    public Task<BackfillWindow?> GetNextWindowAsync(string requestId);
    public Task MarkWindowCompletedAsync(string requestId, string windowId);
    public Task<BackfillStatus> GetStatusAsync(string requestId);
}

BackfillRequest

Circuit Breaker

Automatic failure detection and isolation:

// Configuration on ScheduleDefinition
public int CircuitBreakerThreshold { get; set; } = 5;       // Failures before trip
public int CircuitBreakerCooldownSeconds { get; set; } = 300; // 5 minute cooldown

// Automatic behavior
// 1. Increment ConsecutiveFailures on each failure
// 2. When >= Threshold, set CircuitBreakerTripped = true
// 3. Block all triggers while tripped
// 4. After Cooldown, allow one test run
// 5. If test succeeds, reset; if fails, extend cooldown

Reset

await scheduler.ResetCircuitBreakerAsync("schedule-id", resetBy: "admin");

Dependency Graph

Schedule dependencies with cycle detection:

// Register dependency
_deps.RegisterDependency(
    scheduleId: "daily-report",
    dependsOn: new[] { "load-customers", "load-orders" },
    condition: "ALL_SUCCESS"); // or "ANY_SUCCESS"

// Set max wait
_deps.SetMaxWait("daily-report", maxWaitSeconds: 3600); // 1 hour

// Cycle detection at startup
var cycles = _deps.DetectCycles();
if (cycles.Count > 0)
    Console.WriteLine($"Cycles: {string.Join(" → ", cycles)}");

Dependency Conditions

Usage Examples

Basic Scheduling

var host = new SchedulerHost(editor);
await host.StartAsync();

// Create schedule
var schedule = new ScheduleDefinition
{
    Name = "Hourly Sync",
    PipelineId = "customer-sync",
    SchedulerPluginId = "beep.schedule.cron",
    SchedulerConfig = new Dictionary<string, object>
    {
        ["cron"] = "0 * * * *" // Every hour
    },
    Watermark = new WatermarkConfig
    {
        WatermarkType = "datetime",
        WatermarkColumn = "last_modified"
    }
};

await host.Storage.SaveAsync(schedule);
await host.ReloadAsync();

Event-Driven Trigger

var schedule = new ScheduleDefinition
{
    Name = "On New Order",
    PipelineId = "process-order",
    SchedulerPluginId = "beep.schedule.event",
    SchedulerConfig = new Dictionary<string, object>
    {
        ["eventTopic"] = "orders.created",
        ["filter"] = "status == 'confirmed'"
    }
};

Backfill

var backfillId = await host.SubmitBackfillAsync(
    scheduleId: "customer-sync",
    fromValue: "2024-01-01T00:00:00Z",
    toValue: "2024-01-31T23:59:59Z",
    reason: "Initial load",
    requestedBy: "admin",
    windowSizeSeconds: 86400); // 1 day windows

// Monitor progress
var status = await host.Backfills.GetStatusAsync(backfillId);
Console.WriteLine($"Progress: {status.CompletedWindows}/{status.TotalWindows}");

Monitoring Events

host.RunStarted += (s, e) =>
    Console.WriteLine($"Started: {e.PipelineId} ({e.RunId})");

host.RunCompleted += (s, e) =>
    Console.WriteLine($"Completed: {e.PipelineId} in {e.Duration}");

host.RunFailed += (s, e) =>
    Console.WriteLine($"Failed: {e.PipelineId} - {e.ErrorMessage}");

host.CircuitBreakerChanged += (s, e) =>
    Console.WriteLine($"Circuit breaker: {e.ScheduleId} = {e.IsTripped}");

Dependency Chain

// Load raw data
var loadSchedule = new ScheduleDefinition
{
    Name = "Load Raw Data",
    PipelineId = "load-raw",
    SchedulerPluginId = "beep.schedule.cron",
    SchedulerConfig = new Dictionary<string, object> { ["cron"] = "0 2 * * *" }
};

// Transform depends on load
var transformSchedule = new ScheduleDefinition
{
    Name = "Transform Data",
    PipelineId = "transform-data",
    DependsOn = new List<string> { loadSchedule.Id },
    DependencyCondition = "ALL_SUCCESS",
    DependencyMaxWaitSeconds = 3600
};

// Report depends on transform
var reportSchedule = new ScheduleDefinition
{
    Name = "Generate Report",
    PipelineId = "generate-report",
    DependsOn = new List<string> { transformSchedule.Id },
    DependencyCondition = "ALL_SUCCESS"
};