feat(runtime): ClusterId scope resolution + node-scoped driver-spec parse

2026-06-07 03:05:02 -04:00
parent 7bec2fd4db
commit 24796f2c12
2 changed files with 161 additions and 0 deletions
@@ -18,6 +18,27 @@ public sealed record DriverInstanceSpec(
    string DriverConfig,
    string? ClusterId = null);

+/// <summary>How a node should scope a deployment artifact to its own ClusterId.</summary>
+public enum ClusterFilterMode
+{
+    /// <summary>Apply everything (single-cluster / legacy deployments).</summary>
+    None,
+
+    /// <summary>Filter the artifact to the node's own ClusterId.</summary>
+    ScopeTo,
+
+    /// <summary>Apply nothing (node's cluster row not found in a multi-cluster artifact).</summary>
+    Suppress,
+}
+
+/// <summary>Resolved scoping decision for a node against an artifact.</summary>
+/// <param name="Mode">
+///     None = apply everything (single-cluster / legacy); ScopeTo = filter to <paramref name="ClusterId"/>;
+///     Suppress = apply nothing.
+/// </param>
+/// <param name="ClusterId">The node's ClusterId when <paramref name="Mode"/> is ScopeTo; otherwise null.</param>
+public readonly record struct ClusterScope(ClusterFilterMode Mode, string? ClusterId);
+
 public static class DeploymentArtifact
 {
    private static readonly JsonSerializerOptions JsonOptions = new()
@@ -58,6 +79,73 @@ public static class DeploymentArtifact
        }
    }

+    /// <summary>
+    ///     Resolve how a node should scope a deployment artifact to its own ClusterId. Single-cluster
+    ///     (or legacy) artifacts resolve to <see cref="ClusterFilterMode.None"/> so every existing
+    ///     deployment applies unchanged. In a multi-cluster artifact the node's <c>ClusterNode</c> row
+    ///     (matched by <paramref name="nodeId"/>) selects <see cref="ClusterFilterMode.ScopeTo"/> its
+    ///     ClusterId; a missing row resolves to <see cref="ClusterFilterMode.Suppress"/>. Empty /
+    ///     malformed blobs resolve to <see cref="ClusterFilterMode.None"/> (lenient, matching the
+    ///     other parsers).
+    /// </summary>
+    /// <param name="blob">The deployment artifact blob to inspect.</param>
+    /// <param name="nodeId">The node's identity (e.g. "central-1:4053") to match against <c>Nodes</c>.</param>
+    /// <returns>The resolved <see cref="ClusterScope"/> decision for this node.</returns>
+    public static ClusterScope ResolveClusterScope(ReadOnlySpan<byte> blob, string nodeId)
+    {
+        if (blob.IsEmpty) return new ClusterScope(ClusterFilterMode.None, null);
+        try
+        {
+            using var doc = JsonDocument.Parse(blob.ToArray());
+            var root = doc.RootElement;
+            var clusterCount = root.TryGetProperty("Clusters", out var cl) && cl.ValueKind == JsonValueKind.Array
+                ? cl.GetArrayLength() : 0;
+            if (clusterCount <= 1) return new ClusterScope(ClusterFilterMode.None, null);
+
+            string? myCluster = null;
+            if (root.TryGetProperty("Nodes", out var nodes) && nodes.ValueKind == JsonValueKind.Array)
+            {
+                foreach (var el in nodes.EnumerateArray())
+                {
+                    if (el.ValueKind != JsonValueKind.Object) continue;
+                    var nid = el.TryGetProperty("NodeId", out var nEl) ? nEl.GetString() : null;
+                    if (!string.Equals(nid, nodeId, StringComparison.Ordinal)) continue;
+                    myCluster = el.TryGetProperty("ClusterId", out var cEl) ? cEl.GetString() : null;
+                    break;
+                }
+            }
+            return string.IsNullOrWhiteSpace(myCluster)
+                ? new ClusterScope(ClusterFilterMode.Suppress, null)
+                : new ClusterScope(ClusterFilterMode.ScopeTo, myCluster);
+        }
+        catch (JsonException)
+        {
+            return new ClusterScope(ClusterFilterMode.None, null);
+        }
+    }
+
+    /// <summary>
+    ///     Parse a deployment artifact blob into the driver-instance specs a specific node should
+    ///     spawn, scoped to its own ClusterId via <see cref="ResolveClusterScope"/>. Single-cluster /
+    ///     legacy artifacts return every spec; a multi-cluster artifact returns only the matching
+    ///     cluster's specs (or none when the node's row is absent).
+    /// </summary>
+    /// <param name="blob">The deployment artifact blob to parse.</param>
+    /// <param name="nodeId">The node's identity (e.g. "central-1:4053") used to resolve cluster scope.</param>
+    /// <returns>The driver-instance specs this node should spawn.</returns>
+    public static IReadOnlyList<DriverInstanceSpec> ParseDriverInstances(ReadOnlySpan<byte> blob, string nodeId)
+    {
+        var scope = ResolveClusterScope(blob, nodeId);
+        var all = ParseDriverInstances(blob);
+        return scope.Mode switch
+        {
+            ClusterFilterMode.Suppress => Array.Empty<DriverInstanceSpec>(),
+            ClusterFilterMode.ScopeTo => all.Where(
+                s => string.Equals(s.ClusterId, scope.ClusterId, StringComparison.Ordinal)).ToArray(),
+            _ => all,
+        };
+    }
+
    private static DriverInstanceSpec? TryReadSpec(JsonElement el)
    {
        var rowId = el.TryGetProperty("DriverInstanceRowId", out var rowEl)