ProxyCluster

Overview

ProxyCluster composes multiple IProxyDataSource nodes into a single logical datasource. Implements both IProxyCluster (extends IProxyDataSource which extends IDataSource). All routing, failover, hedging, scatter-gather, and affinity are internal — callers use it as a plain IDataSource.

public partial class ProxyCluster : IProxyCluster

• 10 partial class files
• 5 node routing strategies, 4 proxy-level routing strategies inherited
• 2 read modes: SingleNode (default) + ScatterGather
• Built-in health probing, circuit breaking, outlier detection, hedging
• Config persistence via ConnectionProperties in ConfigEditor.DataConnections
• Slow-start weight ramp, replica lag guard, connection multiplexing

Architecture

ProxyCluster (IProxyCluster : IProxyDataSource : IDataSource)
├── _nodes: ConcurrentDictionary<string, ProxyNode>
│   ├── ProxyNode { NodeId, Proxy (IProxyDataSource), Weight, NodeRole, IsAlive }
│   │   ├── IProxyNode contract: 12 properties
│   │   ├── Outlier detection state (error count, window, ejection)
│   │   └── ReplicaLagMs guard
│   ├── _affinityMap: sessionKey → (nodeId, lastUsed)
│   └── _clusterConnectionSemaphore (optional global connection cap)
├── ProxyPolicy (_clusterPolicy) — single source of truth
├── ICircuitStateStore — per-node circuit breaker state
├── IProxyAuditSink — audit trail sink
├── Node Routing Pipeline:
│   │ 1. Drain check (skip draining nodes)
│   │ 2. Health filter (skip unhealthy / ejected)
│   │ 3. Entity affinity override (if configured)
│   │ 4. Routing rules evaluation (first match wins)
│   │ 5. Traffic split (canary percentage)
│   │ 6. Active strategy: LeastConnections / WeightedRoundRobin / etc.
│   │ 7. ConsistentHash ring for session affinity
├── Background: NodeProbeTimer (liveness), AffinityTtlSweep (cleanup)
└── Optional: Hedging + OutlierDetection + FaultInjection

Constructors

// Constructor 1: policy-only (no DME editor)
public ProxyCluster(
    ProxyPolicy clusterPolicy = null,
    ICircuitStateStore circuitStateStore = null,
    IProxyAuditSink auditSink = null);

// Constructor 2: with DME editor for config persistence
public ProxyCluster(
    IDMEEditor editor,
    string clusterName,
    ProxyPolicy clusterPolicy = null,
    ICircuitStateStore circuitStateStore = null,
    IProxyAuditSink auditSink = null);

Both constructors initialize the cluster-wide SemaphoreSlim if ClusterMaxBackendConnections > 0.

Properties & Events

IDataSource Properties

Backward-Compat Knobs

All Events

Node Management

Adding Nodes

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

// Add from saved config (ConnectionProperties)
cluster.AddNode(new ProxyNode(savedConnection, proxyInstance));

Removing Nodes

bool removed = cluster.RemoveNode("replica2");      // immediate
cluster.DrainNode("replica2", timeoutMs: 30_000);    // graceful — waits for in-flight = 0

Slow-Start Weight Ramp

New nodes start with effective weight = 1 and linearly ramp to their configured weight over SlowStartDurationMs. Configured via ProxyPolicy.SlowStartDurationMs (0 = disabled).

Replica Lag Guard

When ProxyPolicy.MaxReplicaLagMs > 0, replicas with replication lag exceeding the threshold are excluded from read routing. Set 0 to disable.

ProxyNode Class

IProxyNode (contract)

public interface IProxyNode
{
    string NodeId { get; }                           // Unique identifier
    ConnectionProperties NodeConfig { get; }          // Persisted config
    IProxyDataSource Proxy { get; }                   // Underlying proxy
    int Weight { get; set; }                          // Routing weight (mutable for slow-start)
    ProxyDataSourceRole NodeRole { get; set; }        // Primary / Replica / Standby
    bool IsAlive { get; }                             // Last probe result
    DateTime LastProbeUtc { get; }                    // Last probe timestamp
    int InFlightCount { get; }                        // Atomic in-flight counter
    DataSourceMetrics Metrics { get; }                // Aggregated node metrics
    DateTime AddedAtUtc { get; }                      // Node creation timestamp
}

ProxyNode (concrete, internal state)

Internal cluster properties used by probing/routing:

• IsDraining — When true, node receives no new requests
• ConsecutiveFailures / ConsecutiveSuccesses — Probe state
• OutlierErrorCount, OutlierWindowRequests, OutlierWindowErrors — Outlier detection state
• OutlierEjectionCount, OutlierEjectedUntil — Exponential backoff ejection
• ReplicaLagMs — Most recent replication lag measurement

Config Persistence

// Serialize node to ConnectionProperties
ConnectionProperties cfg = node.ToConnectionProperties(
    clusterName: "OrderCluster",
    remoteUrl: "http://worker-a:5100",
    apiKey: "sk-xxx",
    timeoutSeconds: 30);
// Stores: NodeRole, Weight, ClusterName in ParameterList

Node Routing Strategies

ProxyNodeRoutingStrategy (cluster tier)

The cluster also inherits 4 ProxyRoutingStrategy algorithms from IProxyDataSource: WeightedLatency, LeastOutstandingRequests, RoundRobin, HealthWeighted.

Routing Pipeline Order

On every operation, candidate nodes pass through this filter chain:

1. Drain filter — skip draining nodes
2. Health filter — skip unhealthy (probe) + ejected (outlier) + open-circuit nodes
3. Entity affinity — if entity is mapped to a specific node
4. Routing rules — first matching ProxyRoutingRule wins
5. Traffic split — canary percentage routing
6. Active strategy — NodeRoutingStrategy or RoutingStrategy
7. ConsistentHash — when session affinity is active

Declarative Routing Rules

ProxyRoutingRule entries are evaluated in descending Priority order before the normal router. First match wins.

var policy = new ProxyPolicy
{
    RoutingRules = new List<ProxyRoutingRule>
    {
        new ProxyRoutingRule
        {
            OperationPattern = "GetEntity",           // Regex on operation name
            EntityPattern = "Orders",                 // Regex on entity name
            TargetNodeId = "orders-primary",          // Pin to specific node
            Priority = 10
        },
        new ProxyRoutingRule
        {
            EntityPattern = "Logs|Analytics",
            OverrideStrategy = ProxyRoutingOverrideStrategy.RouteToReplica,
            Priority = 5
        }
    }
};

ProxyRoutingOverrideStrategy: RouteToReplica | RouteToPrimary | RouteToAny

Traffic Splits (Canary Deployments)

var policy = new ProxyPolicy
{
    TrafficSplits = new List<TrafficSplitRule>
    {
        new TrafficSplitRule
        {
            TargetNodeId = "canary-node",
            WeightPercent = 10,                       // 10% of traffic
            OperationScope = ProxySplitScope.ReadsOnly // ReadsOnly | WritesOnly | All
        }
    }
};

Session Affinity

// Configure
var policy = new ProxyPolicy
{
    EnableNodeAffinity = true,
    NodeAffinityTtlSeconds = 300    // 5 min session TTL
};

// Affinity is automatic when ProxyExecutionContext.SessionKey is set
// Internal map: ConcurrentDictionary<string, (string NodeId, DateTime LastUsed)>

Background sweep removes expired sessions. Rebalanced when topology changes (events: OnAffinityRebalanced).

Entity Affinity

Hard-pins specific entity names to specific nodes, evaluated BEFORE routing rules in the pipeline.

var policy = new ProxyPolicy
{
    EntityAffinity = new EntityAffinityMap
    {
        Affinities = new()
        {
            ["Products"] = "products-node",
            ["Inventory"] = "inventory-node"
        }
    },
    AffinityFallback = EntityAffinityFallback.ThrowException  // RouteToAny | ThrowException | WaitForOwner
};

Scatter-Gather Reads

When ReadMode = ProxyReadMode.ScatterGather, read operations broadcast to ALL live nodes in parallel and merge results.

var policy = new ProxyPolicy
{
    ReadMode = ProxyReadMode.ScatterGather
};

// Every GetEntity/RunQuery goes to all live nodes simultaneously
// Results from all nodes are merged before returning to caller

Hedged Requests

When a node is slow to respond, issue a parallel request to another node. First response wins.

var policy = new ProxyPolicy
{
    EnableHedging = true,
    HedgingThresholdMs = 100,     // If no response in 100ms, hedge
    MaxHedgeRequests = 2          // Max 2 concurrent copies (incl. original)
};

// Hedge requests use CancellationToken, other copies are cancelled on first success

Outlier Detection (Envoy-style)

Ejects misbehaving nodes with exponentially increasing backoff when OutlierDetectionPolicy is configured.

var policy = new ProxyPolicy
{
    OutlierDetection = new OutlierDetectionPolicy
    {
        ConsecutiveErrorThreshold = 5,     // Eject after 5 consecutive errors
        ErrorRateThreshold = 0.5,          // or 50% error rate over IntervalMs
        IntervalMs = 10_000,               // Analysis window (10s)
        BaseEjectionTimeMs = 30_000,       // First ejection: 30s
        MaxEjectionTimeMs = 300_000,       // Exponential cap: 5 min
        MaxEjectionPercent = 50            // At most 50% of nodes ejected
    }
};

Fault Injection (Chaos Engineering)

Deliberate error and latency injection. Must only be used in test/staging.

var policy = new ProxyPolicy
{
    FaultInjection = new FaultInjectionPolicy
    {
        ErrorRate = 0.05,                // 5% of requests get fake error
        DelayRate = 0.10,                // 10% of requests get artificial delay
        DelayMs = 500,                   // 500ms delay
        TargetNodeId = "fragile-node",   // Limit to specific node (null = all)
        TargetEntity = "Orders"          // Limit to specific entity (null = all)
    }
};

Per-Node Rate Limiting

var policy = new ProxyPolicy
{
    NodeRateLimits = new List<NodeRateLimit>
    {
        new NodeRateLimit
        {
            NodeId = "limited-node",
            MaxRps = 1000,                      // Token-bucket refill rate
            Action = RateLimitAction.Reject      // RouteElsewhere | Queue | Reject
        }
    }
};

Circuit Breaker

Inherited from IProxyDataSource. Per-node circuit breakers managed by ICircuitStateStore.

// Cluster uses the same ICircuitStateStore as ProxyDataSource
var cluster = new ProxyCluster(
    circuitStateStore: new RedisCircuitStateStore(redisConnection));

// State transitions per node:
// Closed → Open (failures ≥ threshold) → HalfOpen (timeout) → Closed (successes ≥ threshold)

Connection Multiplexing

Global cap on backend connections across all nodes via SemaphoreSlim.

var policy = new ProxyPolicy
{
    ClusterMaxBackendConnections = 100,     // 0 = unlimited (default)
    ClusterMaxQueueDepth = 100,             // Max queued when at cap
    ClusterQueueTimeoutMs = 5_000           // Queue timeout
};

Config Persistence

// Constructor with editor enables persistence
var cluster = new ProxyCluster(editor, "CustomerCluster", policy);

cluster.AddNode(new ProxyNode("primary", proxy));
cluster.AddNode(new ProxyNode("replica", replicaProxy, weight: 1, role: Replica));

// Save all nodes to ConfigEditor.DataConnections
cluster.SaveNodesToConfig();

// Reload from config - REQUIRES a factory that turns a stored
// ConnectionProperties record back into a live node proxy.
cluster.LoadLocalNodesFromConfig(conn =>
{
    // return null to skip a record you don't recognise
    return BuildProxyForConnection(conn);
});

Using ProxyCluster as an IDataSource

Because IProxyCluster extends IProxyDataSource, which extends IDataSource, a ProxyCluster is itself a datasource. Everything that consumes IDataSource — UnitofWork<T>, ETL, the sync and import managers — can point at a cluster and get routing, failover and load-balancing for free, without knowing a cluster is involved.

public partial class ProxyCluster : IProxyCluster
// IProxyCluster : IProxyDataSource  ->  IProxyDataSource : IDataSource, IDisposable

Connection lifecycle

public ConnectionState Openconnection();
public ConnectionState Closeconnection();
public void Dispose();

Metadata

public IEnumerable<string> GetEntitesList();
public bool                CheckEntityExist(string entityName);
public int                 GetEntityIdx(string entityName);
public Type                GetEntityType(string entityName);
public EntityStructure     GetEntityStructure(string entityName, bool refresh);
public EntityStructure     GetEntityStructure(EntityStructure fnd, bool refresh = false);
public IEnumerable<ChildRelation>    GetChildTablesList(...);
public IEnumerable<RelationShipKeys> GetEntityforeignkeys(...);

Reads — routed to a selected node

public IEnumerable<object> GetEntity(string entityName, List<AppFilter> filter);
public PagedResult         GetEntity(string entityName, List<AppFilter> filter,
                                     int pageNumber, int pageSize);
public Task<IEnumerable<object>> GetEntityAsync(string entityName, List<AppFilter> filter);
public IEnumerable<object> RunQuery(string qrystr);
public double              GetScalar(string query);
public Task<double>        GetScalarAsync(string query);

Writes — steered to a primary

public IErrorsInfo InsertEntity(string entityName, object insertedData);
public IErrorsInfo UpdateEntity(string entityName, object uploadDataRow);
public IErrorsInfo UpdateEntities(string entityName, object uploadData,
                                  IProgress<PassedArgs> progress);
public IErrorsInfo DeleteEntity(string entityName, object uploadDataRow);
public IErrorsInfo ExecuteSql(string sql);

Schema & scripting

public bool        CreateEntityAs(EntityStructure entity);
public IErrorsInfo CreateEntities(List<EntityStructure> entities);
public IErrorsInfo RunScript(ETLScriptDet dDLScripts);
public IEnumerable<ETLScriptDet> GetCreateEntityScript(List<EntityStructure> entities = null);

Transactions

public IErrorsInfo BeginTransaction(PassedArgs args);
public IErrorsInfo EndTransaction(PassedArgs args);
public IErrorsInfo Commit(PassedArgs args);

Identity defaults

Note what a cluster reports about itself — these are the class defaults, not derived from the nodes:

DatasourceType.Other means you cannot identify a cluster by type at runtime, and Category is hard-coded to RDBMS even when the backing nodes are not relational. Test the concrete type instead:

if (ds is ProxyCluster cluster) { /* cluster-specific work */ }

Example — UnitofWork over a cluster

// The cluster is just an IDataSource as far as UnitofWork is concerned
var uow = new UnitofWork<Order>(dmEditor, cluster, "Orders");

uow.Get(new List<AppFilter>
{
    new AppFilter { FieldName = "Status", Operator = "=", FilterValue = "Open" }
});   // read -> routed to a replica

var order = uow.New();
order.CustomerId = 42;
uow.Add(order);
uow.Commit();   // write -> steered to a primary

Example — arbitrary work with load balancing

var count = await cluster.ExecuteWithLoadBalancing(async ds =>
{
    // 'ds' is the node the cluster selected for this call
    return await ds.GetScalarAsync("SELECT COUNT(*) FROM Orders");
}, isWrite: false);

Full API Reference

Node Management

Connections & Cache

Policy, Metrics & Config

Watchdog / Node Probing

The watchdog is where node health lives. There is no SetNodeHealthy/SetNodeUnhealthy override — health is derived from probe results against the failure/recovery thresholds above.

Execution & Lifetime

ExecuteWithLoadBalancing<T> is the escape hatch: it gives you the selected IDataSource so you can run provider-specific work while still getting routing, the cluster-wide connection cap, retries and circuit-breaking.

All Enum Values

OutlierDetectionPolicy Properties

FaultInjectionPolicy Properties

NodeRateLimit Properties

Source File Map

Usage Examples

Basic Cluster Setup

var policy = new ProxyPolicy
{
    NodeRoutingStrategy = ProxyNodeRoutingStrategy.WeightedRoundRobin,
    Resilience = ProxyResilienceProfile.Balanced
};

var cluster = new ProxyCluster(policy);

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

// Use as IDataSource
IDataSource ds = cluster;
var data = ds.GetEntity("Customers", filters);

Scatter-Gather with Hedging

var policy = new ProxyPolicy
{
    ReadMode = ProxyReadMode.ScatterGather,
    EnableHedging = true,
    HedgingThresholdMs = 50,
    MaxHedgeRequests = 3
};
var cluster = new ProxyCluster(policy);
// Reads broadcast to all nodes. Slow nodes get hedged.

Routing Rules + Traffic Split

var policy = new ProxyPolicy
{
    RoutingRules = new List<ProxyRoutingRule>
    {
        new() { EntityPattern = "Orders", TargetNodeId = "orders-primary", Priority = 10 }
    },
    TrafficSplits = new List<TrafficSplitRule>
    {
        new() { TargetNodeId = "canary", WeightPercent = 5, OperationScope = ProxySplitScope.All }
    }
};

With Config Persistence

var cluster = new ProxyCluster(editor, "CustomerCluster", policy);
cluster.AddNode(new ProxyNode("node-a", proxyA));
cluster.AddNode(new ProxyNode("node-b", proxyB));
cluster.SaveNodesToConfig();          // Persists to ConfigEditor.DataConnections

// Later session
var restored = new ProxyCluster(editor, "CustomerCluster", policy);
restored.LoadLocalNodesFromConfig();  // Reloads nodes from config

Graceful Drain Before Deploy

// Drain node before taking it offline
cluster.DrainNode("replica2", timeoutMs: 60_000);
cluster.RemoveNode("replica2");

// Async version
await cluster.DrainNodeAsync("replica2", timeoutMs: 60_000, ct);

Monitoring

cluster.OnNodeDown += (s, e) =>
    Console.WriteLine($"Node down: {e.NodeId}");
cluster.OnNodeRestored += (s, e) =>
    Console.WriteLine($"Node restored: {e.NodeId}");
cluster.OnFailover += (s, e) =>
    Console.WriteLine($"Failover: {e.FromDataSource} → {e.ToDataSource}");

// SLO snapshots
foreach (var snap in cluster.GetClusterSloSnapshots())
    Console.WriteLine($"{snap.DataSourceName}: p99={snap.P99LatencyMs:F1}ms");