ProxyDataSource

Overview

ProxyDataSource wraps one or more concrete IDataSource instances behind a single, transparent facade that the rest of the application treats as an ordinary data source. It adds the following capabilities without any changes to calling code:

Architecture & Phase Map

The implementation is split across partial classes by concern, each corresponding to a numbered phase of the design plan:

Request lifecycle

caller → ProxyDataSource.GetEntity(...)
            │
            ├─ SelectCandidates()          [Routing]
            │     filter healthy + closed circuits
            │     order by routing strategy
            │
            ├─ for each candidate:
            │     GetPooledConnection()
            │     for 1..MaxRetries:
            │         execute operation
            │         ├─ success → RecordSuccess, emit audit, return
            │         └─ failure → ProxyErrorClassifier.Classify()
            │                       ├─ non-retryable → throw immediately
            │                       ├─ saturation    → 2× back-off, retry
            │                       └─ transient     → exp back-off+jitter, retry
            │
            └─ all candidates exhausted → log + throw last exception

Core Types & Enums

ProxyDataSourceRole

ProxyRoutingStrategy

ProxyOperationSafety

DataSourceMetrics

Thread-safe counters maintained per backend datasource. Available via GetMetrics().

ProxySloSnapshot

Point-in-time SLO report returned by GetSloSnapshot(dsName).

PooledConnection

A lease on a connection inside the per-source pool.

Constructors

Preferred constructor — policy-based

public ProxyDataSource(
    IDMEEditor         dmeEditor,
    List<string>       dataSourceNames,
    ProxyPolicy        policy           = null,
    ICircuitStateStore circuitStateStore = null,
    IProxyAuditSink    auditSink        = null)

Backwards-compatible constructor

public ProxyDataSource(
    IDMEEditor   dmeEditor,
    List<string> dataSourceNames,
    int?         maxRetries              = null,
    int?         retryDelayMilliseconds  = null,
    int?         healthCheckIntervalMs   = null)

Wraps the scalar parameters into a ProxyPolicy internally. Provided for compatibility with older call sites; prefer the policy overload for new code.

ProxyPolicy

ProxyPolicy is the single source of truth for every tunable behaviour. It is applied at construction time and can be swapped at runtime via ApplyPolicy(ProxyPolicy) without recreating the proxy.

Top-level properties

Runtime hot-swap

// Change to aggressive failover without recreating the proxy
var aggressivePolicy = new ProxyPolicy
{
    Name            = "HighAvailability",
    RoutingStrategy = ProxyRoutingStrategy.HealthWeighted,
    Resilience      = ProxyResilienceProfile.AggressiveFailover,
    HealthCheckIntervalMs = 10_000
};

proxy.ApplyPolicy(aggressivePolicy);   // thread-safe; stops old timer, reinitialises circuits

Resilience Profiles

ProxyResilienceProfile bundles all retry and circuit-breaker knobs. Three built-in presets cover most workloads; use Custom for fine tuning.

// Use a preset
var policy = new ProxyPolicy { Resilience = ProxyResilienceProfile.Conservative };

// Or inline-custom
var policy = new ProxyPolicy
{
    Resilience = new ProxyResilienceProfile
    {
        ProfileType              = ProxyResilienceProfileType.Custom,
        MaxRetries               = 4,
        RetryBaseDelayMs         = 300,
        FailureThreshold         = 7,
        CircuitResetTimeout      = TimeSpan.FromMinutes(2),
        ConsecutiveSuccessesToClose = 2
    }
};

Routing Strategies

On every read operation, SelectCandidates() builds an ordered list of healthy (non-open-circuit) sources and passes it to the active routing algorithm. The execution loop tries candidates in that order with retries. Writes are pre-filtered to Primary-role sources only.

Circuit Breaker

Each registered datasource gets its own CircuitBreaker instance managed by the active ICircuitStateStore. The breaker uses three states:

Severity-weighted failure accumulation

Swappable state store

The ICircuitStateStore interface allows replacing the default in-process store with a distributed backend so circuit state is shared across multiple proxy instances:

public interface ICircuitStateStore
{
    bool CanExecute(string dsName);
    void RecordSuccess(string dsName);
    void RecordFailure(string dsName, ProxyErrorSeverity severity = ProxyErrorSeverity.Medium);
    void Initialize(string dsName, int failureThreshold, TimeSpan resetTimeout, int successThreshold = 2);
    void Remove(string dsName);
}

Use the provided ProxyRedisCircuitStateStore for clusters that share state across processes, or implement the interface against any storage backend.

Connection Pool

Each registered datasource has its own bounded ConcurrentQueue<PooledConnection>. Connections are leased from the pool before each operation and returned after. Idle eviction runs on the health-check timer interval.

Pool API

// Internal — used by execution helpers
IDataSource lease  = proxy.GetPooledConnection(dsName);   // borrow from pool (or open new)
proxy.ReturnConnection(dsName, lease);                    // return lease after use
IDataSource direct = proxy.GetConnection(dsName);         // bypass pool — new connection always

Caching

Enabled via ProxyCacheProfile inside ProxyPolicy. The cache is backed by a CacheScope from the BeepDM caching subsystem. The proxy uses a stale-while-revalidate strategy: a soft TTL (the visible expiry) and a hard TTL (stale extension window). When a key is between soft and hard TTL the stale value is returned immediately while a background refresh populates the cache.

Cache API

// Read with cache (respects stale-while-revalidate)
var rows = proxy.GetEntityWithCache("Orders", filters, expiration: TimeSpan.FromMinutes(5));

// Manual invalidation
proxy.InvalidateCache("Orders");    // invalidate one entity
proxy.InvalidateCache();            // clear all cached entries

Watchdog

The watchdog runs as a background Task on a tight cycle (default 5 s) and probes every registered datasource with a lightweight liveness call capped at WatchdogProbeTimeoutMs (default 2 s). It maintains independent consecutive failure/success counters per source, separate from the main circuit breaker counters.

Role management

1. When a Primary source fails WatchdogFailureThreshold consecutive probes, the watchdog promotes the highest-weighted healthy Replica to Primary.
2. The promotion is recorded in _originalRoles so the role can be reversed.
3. When the original Primary recovers (WatchdogRecoveryThreshold consecutive successes), the watchdog demotes it back to Replica and restores the promoted Replica.

Knobs & events

Usage

proxy.WatchdogIntervalMs       = 3_000;  // tighter probe on critical DB
proxy.WatchdogFailureThreshold = 3;

proxy.OnRolePromoted += (_, e) =>
    Console.WriteLine($"{e.DataSourceName} promoted to {e.NewRole}");

proxy.StartWatchdog();

// ... application lifetime ...

proxy.StopWatchdog();  // graceful — waits up to 5 s for loop exit

Observability & SLO

The observability layer maintains a bounded 500-sample circular latency buffer per datasource and exposes p50/p95/p99 percentile summaries through the SLO snapshot API. Samples are recorded automatically by the execution helpers on every successful call.

SLO snapshot API

// Single source snapshot
ProxySloSnapshot snap = proxy.GetSloSnapshot("orders-primary");
Console.WriteLine($"p99={snap.P99LatencyMs:F1}ms  err={snap.ErrorRatePercent:F2}%  cache={snap.CacheHitRatio:P}");

// All sources at once
IReadOnlyList<ProxySloSnapshot> all = proxy.GetAllSloSnapshots();
foreach (var s in all)
    Console.WriteLine($"{s.DataSourceName}: p95={s.P95LatencyMs:F1}ms");

Raw metrics access

IDictionary<string, DataSourceMetrics> metrics = proxy.GetMetrics();
foreach (var (name, m) in metrics)
    Console.WriteLine($"{name}: total={m.TotalRequests}  failed={m.FailedRequests}  breaks={m.CircuitBreaks}");

Audit

Every proxied operation emits a ProxyAuditEntry to the configured IProxyAuditSink after the operation completes — whether it succeeded or failed.

ProxyAuditEntry fields

Provided sinks

// Enable file audit sink at construction
var sink = new FileProxyAuditSink(directory: @"C:\Logs\ProxyAudit", queueCapacity: 10_000);

var proxy = new ProxyDataSource(
    dmeEditor       : editor,
    dataSourceNames : new List<string> { "db1", "db2" },
    auditSink       : sink);

// ... use proxy ...

sink.Dispose();   // flushes background queue, waits up to 5 s

Error Taxonomy

ProxyErrorClassifier.Classify(Exception) maps every exception to a category + severity pair that drives retry eligibility and circuit-breaker weight:

Saturation is special: the proxy backs off with a delay proportional to attempt × 2 (double the normal back-off) to reduce load on an overwhelmed backend rather than hammering it with rapid retries.

ProxyCluster

ProxyCluster implements IProxyCluster : IProxyDataSource and represents a cluster of ProxyNode instances behind a single IDataSource facade. Use it when you need a second tier of abstraction: clusters of proxies rather than clusters of bare datasources.

Node management

var cluster = new ProxyCluster(
    clusterPolicy : new ProxyPolicy { RoutingStrategy = ProxyRoutingStrategy.LeastOutstandingRequests },
    auditSink     : new FileProxyAuditSink(@"C:\Logs\Cluster"));

// Add nodes
cluster.AddNode(new ProxyNode("primary",  primaryProxy,  weight: 2, role: ProxyDataSourceRole.Primary));
cluster.AddNode(new ProxyNode("replica1", replicaProxy1, weight: 1, role: ProxyDataSourceRole.Replica));
cluster.AddNode(new ProxyNode("replica2", replicaProxy2, weight: 1, role: ProxyDataSourceRole.Replica));

// Remove a node gracefully (drains in-flight requests)
cluster.RemoveNode("replica2");

// Inspect nodes
IReadOnlyList<IProxyNode> nodes = cluster.GetNodes();
foreach (var n in nodes)
    Console.WriteLine($"{n.NodeId}: alive={n.IsAlive} weight={n.Weight} inflight={n.InFlightCount}");

ProxyNode properties

Cluster routing strategies (ProxyNodeRoutingStrategy)

Cluster events

Usage Examples

Example 1 — Basic read with default policy

var proxy = new ProxyDataSource(
    dmeEditor       : editor,
    dataSourceNames : new List<string> { "orders-primary", "orders-replica" });

proxy.Openconnection();

var rows = proxy.GetEntity("Orders", new List<AppFilter>
{
    new AppFilter { FieldName = "Status", Operator = "=", FilterValue = "Open" }
});

foreach (var row in rows)
    Console.WriteLine(row);

proxy.Closeconnection();
proxy.Dispose();

Example 2 — Custom policy with caching

var policy = new ProxyPolicy
{
    Name            = "ProductsPolicy",
    RoutingStrategy = ProxyRoutingStrategy.WeightedLatency,
    Resilience = new ProxyResilienceProfile
    {
        ProfileType         = ProxyResilienceProfileType.Custom,
        MaxRetries          = 3,
        RetryBaseDelayMs    = 200,
        FailureThreshold    = 5,
        CircuitResetTimeout = TimeSpan.FromSeconds(30)
    },
    Cache = new ProxyCacheProfile
    {
        Enabled           = true,
        DefaultExpiration = TimeSpan.FromSeconds(60),
        Consistency       = ProxyCacheConsistency.WriteThrough
    }
};

var proxy = new ProxyDataSource(
    dmeEditor       : editor,
    dataSourceNames : new List<string> { "products-db1", "products-db2" },
    policy          : policy);

proxy.Openconnection();
var result = proxy.RunQuery("SELECT COUNT(*) FROM Products");
proxy.Dispose();

Example 3 — Read/write role separation

var proxy = new ProxyDataSource(
    dmeEditor       : editor,
    dataSourceNames : new List<string> { "sales-primary", "sales-replica-1", "sales-replica-2" });

// Assign roles
proxy.SetRole("sales-primary",   ProxyDataSourceRole.Primary);
proxy.SetRole("sales-replica-1", ProxyDataSourceRole.Replica);
proxy.SetRole("sales-replica-2", ProxyDataSourceRole.Replica);

proxy.Openconnection();

// Read — may land on any healthy node
var items = proxy.GetEntity("SaleItems", null);

// Write — guaranteed to go to Primary-role source only
proxy.InsertEntity("Orders", new { OrderId = Guid.NewGuid(), Total = 99.99 });

proxy.Dispose();

Example 4 — Watchdog with role promotion events

var proxy = new ProxyDataSource(
    dmeEditor       : editor,
    dataSourceNames : new List<string> { "db-primary", "db-replica" });

proxy.SetRole("db-primary", ProxyDataSourceRole.Primary);
proxy.SetRole("db-replica", ProxyDataSourceRole.Replica);

proxy.WatchdogIntervalMs       = 3_000;
proxy.WatchdogFailureThreshold = 2;
proxy.WatchdogRecoveryThreshold = 4;

proxy.OnRolePromoted += (_, e) =>
    logger.LogWarning($"PROMOTED: {e.DataSourceName} is now {e.NewRole}");
proxy.OnRoleDemoted += (_, e) =>
    logger.LogInformation($"DEMOTED: {e.DataSourceName} restored to {e.NewRole}");

proxy.Openconnection();
proxy.StartWatchdog();

// ... application lifetime ...

proxy.StopWatchdog();
proxy.Dispose();

Example 5 — Audit + observability

using var sink = new FileProxyAuditSink(@"C:\Logs\ProxyAudit");

var proxy = new ProxyDataSource(
    dmeEditor       : editor,
    dataSourceNames : new List<string> { "analytics-db1", "analytics-db2" },
    auditSink       : sink);

proxy.Openconnection();

// ... run operations ...

// Check SLO
foreach (var snap in proxy.GetAllSloSnapshots())
{
    Console.WriteLine(
        $"{snap.DataSourceName}: p50={snap.P50LatencyMs:F1}ms  " +
        $"p99={snap.P99LatencyMs:F1}ms  " +
        $"err={snap.ErrorRatePercent:F2}%  " +
        $"cache={snap.CacheHitRatio:P}");
}

proxy.Dispose();
// sink.Dispose() called by using — flushes queue

Example 6 — ProxyCluster

// Build individual proxies (one per physical server)
var primary = new ProxyDataSource(editor, new List<string> { "server-a" });
var replica = new ProxyDataSource(editor, new List<string> { "server-b" });

// Compose into a cluster
var cluster = new ProxyCluster(
    clusterPolicy : new ProxyPolicy { RoutingStrategy = ProxyRoutingStrategy.LeastOutstandingRequests });

cluster.AddNode(new ProxyNode("primary",  primary,  weight: 2, role: ProxyDataSourceRole.Primary));
cluster.AddNode(new ProxyNode("replica1", replica,  weight: 1, role: ProxyDataSourceRole.Replica));

cluster.OnNodeDown      += (_, e) => Console.WriteLine($"Node down: {e.NodeId}");
cluster.OnNodeRestored  += (_, e) => Console.WriteLine($"Node up:   {e.NodeId}");

// Use the cluster exactly like a plain IDataSource
IDataSource ds = cluster;
var data = ds.GetEntity("Products", null);

Example 7 — Hot policy swap

// Start with balanced policy
var proxy = new ProxyDataSource(editor, new List<string> { "db1", "db2" });

// During an incident: switch to aggressive failover in place
proxy.ApplyPolicy(new ProxyPolicy
{
    Name            = "Incident-2024",
    RoutingStrategy = ProxyRoutingStrategy.HealthWeighted,
    Resilience      = ProxyResilienceProfile.AggressiveFailover,
    HealthCheckIntervalMs = 5_000
});

// After the incident: restore normal policy
proxy.ApplyPolicy(new ProxyPolicy { Resilience = ProxyResilienceProfile.Balanced });

Full API Reference

IProxyDataSource (complete interface)

public interface IProxyDataSource : IDataSource, IDisposable
{
    // Events
    event EventHandler<FailoverEventArgs> OnFailover;
    event EventHandler<RecoveryEventArgs> OnRecovery;

    // Policy
    void ApplyPolicy(ProxyPolicy policy);

    // Backward-compat knobs
    int MaxRetries { get; set; }
    int RetryDelayMilliseconds { get; set; }
    int HealthCheckIntervalMilliseconds  { get; set; }

    // Load-balanced execution
    Task<T> ExecuteWithLoadBalancing<T>(Func<IDataSource, Task<T>> operation, bool isWrite = false, CancellationToken cancellationToken = default);

    // Connection pool
    IDataSource GetPooledConnection(string dsName);
    void ReturnConnection(string dsName, IDataSource connection);
    IDataSource GetConnection(string dsName);

    // Cache
    object GetEntityWithCache(string entityName, List<AppFilter> filter, TimeSpan? expiration = null);
    void InvalidateCache(string entityName = null);

    // Metrics & SLO
    IDictionary<string, DataSourceMetrics> GetMetrics();
    ProxySloSnapshot GetSloSnapshot(string dsName);
    IReadOnlyList<ProxySloSnapshot> GetAllSloSnapshots();

    // Datasource membership
    void AddDataSource(string dsName, int weight = 1);
    void RemoveDataSource(string dsName);
    void SetRole(string dsName, ProxyDataSourceRole role);

    // Watchdog
    int WatchdogIntervalMs { get; set; }
    int WatchdogProbeTimeoutMs { get; set; }
    int WatchdogFailureThreshold { get; set; }
    int WatchdogRecoveryThreshold { get; set; }
    event EventHandler<RoleChangeEventArgs> OnRolePromoted;
    event EventHandler<RoleChangeEventArgs> OnRoleDemoted;
    void StartWatchdog();
    void StopWatchdog();
    IReadOnlyList<WatchdogNodeStatus> GetWatchdogStatus();

    // Audit
    IProxyAuditSink AuditSink { get; set; }
}

IProxyCluster (complete interface)

public interface IProxyCluster : IProxyDataSource
{
    // Node management
    void AddNode(IProxyNode node);
    void RemoveNode(string nodeId);
    IReadOnlyList<IProxyNode> GetNodes();

    // Drain / rolling restart
    void DrainNode(string nodeId, int timeoutMs = 30_000);
    Task DrainNodeAsync(string nodeId, int timeoutMs = 30_000, CancellationToken ct = default);

    // Policy propagation
    void ApplyClusterPolicy(ProxyPolicy policy);

    // Aggregate observability
    IDictionary<string, DataSourceMetrics> GetClusterMetrics();
    IReadOnlyList<ProxySloSnapshot> GetClusterSloSnapshots();

    // Cluster events
    event EventHandler<NodeStatusEventArgs>         OnNodeDown;
    event EventHandler<NodeStatusEventArgs>         OnNodeRestored;
    event EventHandler<RoleChangeEventArgs>         OnNodePromoted;
    event EventHandler<RoleChangeEventArgs>         OnNodeDemoted;
    event EventHandler<ClusterPolicyChangedEventArgs> OnClusterPolicyChanged;
    event EventHandler<AffinityRebalancedEventArgs> OnAffinityRebalanced;

    // Distributed transaction
    bool SupportsTwoPhaseCommit => false;
}

IProxyNode (Phase 11)

public interface IProxyNode
{
    string NodeId { get; }
    int Weight { get; set; }
    ProxyDataSourceRole NodeRole { get; set; }
    bool IsAlive { get; }
    int InFlightCount { get; }
    DataSourceMetrics Metrics { get; }
    ConnectionProperties NodeConfig { get; set; }
}

ProxyPolicy — Complete Properties

ICircuitStateStore (complete)

public interface ICircuitStateStore
{
    bool CanExecute(string dsName);
    void RecordSuccess(string dsName);
    void RecordFailure(string dsName, ProxyErrorSeverity severity = ProxyErrorSeverity.Medium);
    void Initialize(string dsName, int failureThreshold, TimeSpan resetTimeout, int successThreshold = 2);
    void Remove(string dsName);
}

IProxyAuditSink & IProxyAuditEntry

public interface IProxyAuditSink
{
    /// Must not throw and must complete quickly
    void Write(IProxyAuditEntry entry);
}

public interface IProxyAuditEntry
{
    string             CorrelationId  { get; }
    string             OperationName { get; }
    string             SelectedSource { get; }
    bool               Succeeded     { get; }
    int                TotalAttempts { get; }
    long               ElapsedMs    { get; }
    ProxyOperationSafety Safety       { get; }
    IReadOnlyList<ProxyAttemptRecord> Attempts { get; }
}

IProxyNode

public interface IProxyNode
{
    string NodeId { get; }
    int Weight { get; set; }
    ProxyDataSourceRole NodeRole { get; set; }
    bool IsAlive { get; }
    int InFlightCount { get; }
    DataSourceMetrics Metrics { get; }
    ConnectionProperties NodeConfig { get; set; }
}

EntityAffinityMap

File: EntityAffinityMap.cs. Binds entity names to specific cluster nodes.

public class EntityAffinityMap
{
    public Dictionary<string, string> Affinities { get; init; } = new();
}

Fallback behavior: RouteToAny | ThrowException | WaitForOwner.

ProxyLogRedactor

File: ProxyLogRedactor.cs. Scrubs PII (passwords, tokens, connection strings) from proxy log messages when EnableLogRedaction = true.

All Enums Reference

Class: CircuitBreaker

Thread-safe three-state circuit breaker with severity-weighted failure accumulation. Critical severity immediately breaches the threshold; High counts as 2 failures. Uses separate _successThreshold for HalfOpen → Closed transitions.

public class CircuitBreaker
{
    public CircuitBreakerState State { get; }
    public int FailureCount { get; }
    public DateTime LastStateChange { get; }

    public CircuitBreaker(int failureThreshold, TimeSpan resetTimeout, int successThreshold = 2);

    public bool CanExecute();                           // Test + transition Open→HalfOpen
    public void RecordSuccess();                        // HalfOpen successes → Closed
    public void RecordFailure(ProxyErrorSeverity severity = Medium);
    public void Reset();                                // Force Closed
}

Source Files Reference