Add graph reachability example, e2e test, and clarify paper assumptions

Author: cristianoc

examples/LiveHarness.res

@@ -45,6 +45,9 @@ module Client = {
   let updateInput = (broker, entries) =>
     SkipruntimeHelpers.update(broker, "numbers", entries)
 
+  let updateEdges = (broker, entries) =>
+    SkipruntimeHelpers.update(broker, "edges", entries)
+
   let getStreamUrl = async (opts: SkipruntimeServer.runOptions, broker, resource) => {
     let uuid = await SkipruntimeHelpers.getStreamUUID(broker, resource, None)
     `http://${localhost}:${opts.streaming_port->Int.toString}/v1/streams/${uuid}`
@@ -138,16 +141,30 @@ let run = async () => {
   await Client.snapshot(broker, "numbers", "harness: numbers")
   await Client.snapshot(broker, "doubled", "harness: doubled")
   await Client.snapshot(broker, "sum", "harness: sum")
+  await Client.snapshot(broker, "dead", "harness: dead code (unreachable nodes)")
   Console.log2("harness: counters after initial snapshot", Server.getRunStats())
 
   // Phase 2: Update c from 3 to 5.
   await Client.updateInput(broker, [(JSON.String("c"), [JSON.Number(5.)])])
   await Client.snapshot(broker, "numbers", "harness: numbers after c→5")
   await Client.snapshot(broker, "doubled", "harness: doubled after c→5")
   await Client.snapshot(broker, "sum", "harness: sum after c→5")
+  await Client.snapshot(broker, "dead", "harness: dead code after c→5 (unchanged)")
   Console.log2("harness: client sum after c→5 (from SSE)", ClientSum.getTotal())
   Console.log2("harness: counters after c→5", Server.getRunStats())
 
+  // Phase 3: Remove an edge to create dead code.
+  await Client.updateEdges(broker, [
+    (
+      JSON.String("fileA"),
+      [
+        // drop util -> lib edge to make lib/helper unreachable
+        JSON.Array([JSON.String("main"), JSON.String("util")]),
+      ],
+    ),
+  ])
+  await Client.snapshot(broker, "dead", "harness: dead code after dropping util→lib")
+
   ClientSum.close()
   await Server.stop(server)
   Console.log("harness: service closed")

examples/LiveHarness.res.js

@@ -3,30 +3,198 @@ const log = (...args) => {
     if (ENABLE_LOGS)
         console.log(...args);
 };
+// Utilities for graph state manipulation (JSON-friendly in the accumulator; Sets/Maps for computation).
+const toSets = (state) => {
+    const nodes = new Set(state.nodes);
+    const roots = new Set(state.roots);
+    const reachable = new Set(state.reachable);
+    const edgesOut = new Map();
+    for (const [k, vs] of Object.entries(state.edgesOut)) {
+        edgesOut.set(k, new Set(vs));
+    }
+    const incoming = new Map();
+    for (const [k, v] of Object.entries(state.incoming)) {
+        incoming.set(k, v);
+    }
+    return { nodes, roots, reachable, edgesOut, incoming };
+};
+const fromSets = (s) => ({
+    nodes: Array.from(s.nodes),
+    roots: Array.from(s.roots),
+    reachable: Array.from(s.reachable),
+    edgesOut: Object.fromEntries(Array.from(s.edgesOut.entries()).map(([k, vs]) => [k, Array.from(vs)])),
+    incoming: Object.fromEntries(Array.from(s.incoming.entries())),
+});
+const ensureNode = (s, id) => {
+    s.nodes.add(id);
+    if (!s.edgesOut.has(id))
+        s.edgesOut.set(id, new Set());
+    if (!s.incoming.has(id))
+        s.incoming.set(id, 0);
+};
+const propagateReachable = (s, start) => {
+    const queue = [...start];
+    while (queue.length > 0) {
+        const u = queue.pop();
+        if (s.reachable.has(u))
+            continue;
+        s.reachable.add(u);
+        const outs = s.edgesOut.get(u);
+        if (!outs)
+            continue;
+        for (const v of outs) {
+            ensureNode(s, v);
+            const newCount = (s.incoming.get(v) ?? 0) + 1;
+            s.incoming.set(v, newCount);
+            if (!s.reachable.has(v)) {
+                queue.push(v);
+            }
+        }
+    }
+};
+const propagateUnreachable = (s, start) => {
+    const queue = [...start];
+    while (queue.length > 0) {
+        const u = queue.pop();
+        if (!s.reachable.has(u))
+            continue;
+        if (s.roots.has(u))
+            continue; // roots stay reachable
+        const incoming = s.incoming.get(u) ?? 0;
+        if (incoming > 0)
+            continue;
+        s.reachable.delete(u);
+        const outs = s.edgesOut.get(u);
+        if (!outs)
+            continue;
+        for (const v of outs) {
+            const newCount = Math.max(0, (s.incoming.get(v) ?? 0) - 1);
+            s.incoming.set(v, newCount);
+            if (newCount === 0 && !s.roots.has(v)) {
+                queue.push(v);
+            }
+        }
+    }
+};
+class GraphReducer {
+    constructor() {
+        this.initial = {
+            nodes: [],
+            roots: [],
+            reachable: [],
+            edgesOut: {},
+            incoming: {},
+        };
+    }
+    add(acc, value) {
+        const state = toSets(acc ?? this.initial);
+        if (value.kind === "node") {
+            ensureNode(state, value.id);
+        }
+        else if (value.kind === "root") {
+            ensureNode(state, value.id);
+            if (!state.roots.has(value.id)) {
+                state.roots.add(value.id);
+                propagateReachable(state, [value.id]);
+            }
+        }
+        else {
+            ensureNode(state, value.from);
+            ensureNode(state, value.to);
+            const outs = state.edgesOut.get(value.from);
+            if (!outs.has(value.to)) {
+                outs.add(value.to);
+                if (state.reachable.has(value.from)) {
+                    const newCount = (state.incoming.get(value.to) ?? 0) + 1;
+                    state.incoming.set(value.to, newCount);
+                    if (!state.reachable.has(value.to)) {
+                        propagateReachable(state, [value.to]);
+                    }
+                }
+            }
+        }
+        return fromSets(state);
+    }
+    remove(acc, value) {
+        const state = toSets(acc ?? this.initial);
+        if (value.kind === "node") {
+            // Removing a node: drop from sets; edges incident to it are left intact for simplicity.
+            state.nodes.delete(value.id);
+            state.roots.delete(value.id);
+            if (state.reachable.has(value.id)) {
+                state.reachable.delete(value.id);
+                propagateUnreachable(state, Array.from(state.edgesOut.get(value.id) ?? []));
+            }
+            state.edgesOut.delete(value.id);
+            state.incoming.delete(value.id);
+        }
+        else if (value.kind === "root") {
+            state.roots.delete(value.id);
+            if (state.roots.has(value.id)) {
+                // no-op if still a root via another contribution
+            }
+            else if (state.reachable.has(value.id)) {
+                // Remove reachability if no other incoming
+                if ((state.incoming.get(value.id) ?? 0) === 0) {
+                    propagateUnreachable(state, [value.id]);
+                }
+            }
+        }
+        else {
+            const outs = state.edgesOut.get(value.from);
+            if (outs && outs.has(value.to)) {
+                outs.delete(value.to);
+                if (state.reachable.has(value.from)) {
+                    const newCount = Math.max(0, (state.incoming.get(value.to) ?? 0) - 1);
+                    state.incoming.set(value.to, newCount);
+                    if (newCount === 0 && !state.roots.has(value.to)) {
+                        propagateUnreachable(state, [value.to]);
+                    }
+                }
+            }
+        }
+        return fromSets(state);
+    }
+}
+class EdgeToGraphMapper {
+    mapEntry(_key, values, _ctx) {
+        return values.toArray().map(([from, to]) => ["graph", { kind: "edge", from, to }]);
+    }
+}
+class NodeToGraphMapper {
+    mapEntry(_key, values, _ctx) {
+        return values.toArray().map((id) => ["graph", { kind: "node", id }]);
+    }
+}
+class RootToGraphMapper {
+    mapEntry(_key, values, _ctx) {
+        return values.toArray().map((id) => ["graph", { kind: "root", id }]);
+    }
+}
 // Mapper: multiply numeric values by 2, keep the same key.
 class DoubleMapper {
-    static runs = 0;
     mapEntry(key, values, _ctx) {
         DoubleMapper.runs += 1;
         log("mapper:doubled run", DoubleMapper.runs, "key", key);
         const n = values.getUnique();
         return [[key, n * 2]];
     }
 }
+DoubleMapper.runs = 0;
 // Mapper for sum: emit all values under a single "total" key.
 class TotalMapper {
-    static runs = 0;
     mapEntry(_key, values, _ctx) {
         TotalMapper.runs += 1;
         log("mapper:total run", TotalMapper.runs);
         return values.toArray().map((v) => ["total", v]);
     }
 }
+TotalMapper.runs = 0;
 // Reducer for sum: correctly implements add/remove.
 class SumReducer {
-    static runsAdd = 0;
-    static runsRemove = 0;
-    initial = 0;
+    constructor() {
+        this.initial = 0;
+    }
     add(acc, value) {
         SumReducer.runsAdd += 1;
         log("reducer:sum add", SumReducer.runsAdd);
@@ -38,6 +206,8 @@ class SumReducer {
         return acc - value;
     }
 }
+SumReducer.runsAdd = 0;
+SumReducer.runsRemove = 0;
 class NumbersResource {
     instantiate(collections) {
         return collections.numbers;
@@ -53,6 +223,20 @@ class SumResource {
         return collections.numbers.map(TotalMapper).reduce(SumReducer);
     }
 }
+class DeadNodesResource {
+    instantiate(collections) {
+        // Map graph state to dead-node list: nodes minus reachable.
+        class DeadMapper {
+            mapEntry(key, values, _ctx) {
+                const state = values.getUnique();
+                const reachable = new Set(state.reachable);
+                const dead = state.nodes.filter((n) => !reachable.has(n));
+                return [[key, dead]];
+            }
+        }
+        return collections.graphState.map(DeadMapper);
+    }
+}
 export const service = {
     initialData: {
         numbers: [
@@ -67,13 +251,51 @@ export const service = {
             ["i", [9]],
             ["j", [10]],
         ],
+        edges: [
+            ["fileA", [
+                    ["main", "util"],
+                    ["util", "lib"],
+                ]],
+            ["fileB", [["lib", "helper"]]],
+            ["fileC", []],
+        ],
+        nodes: [
+            ["fileA", ["main", "util"]],
+            ["fileB", ["lib", "helper"]],
+            ["fileC", ["unused"]],
+        ],
+        roots: [
+            ["fileA", ["main"]],
+            ["fileB", []],
+            ["fileC", []],
+        ],
     },
     resources: {
         numbers: NumbersResource,
         doubled: DoubledResource,
         sum: SumResource,
+        dead: DeadNodesResource,
+    },
+    createGraph: (inputs) => {
+        const toGraphEdges = inputs.edges.map(EdgeToGraphMapper);
+        const toGraphNodes = inputs.nodes.map(NodeToGraphMapper);
+        const toGraphRoots = inputs.roots.map(RootToGraphMapper);
+        const graphInputs = toGraphEdges.merge(toGraphNodes, toGraphRoots);
+        const graphState = graphInputs.reduce(GraphReducer);
+        return {
+            ...inputs,
+            graphInputs,
+            graphState,
+            deadNodes: graphState.map(class DeadMapper {
+                mapEntry(key, values, _ctx) {
+                    const state = values.getUnique();
+                    const reachable = new Set(state.reachable);
+                    const dead = state.nodes.filter((n) => !reachable.has(n));
+                    return [[key, dead]];
+                }
+            }),
+        };
     },
-    createGraph: (inputs) => inputs,
 };
 export const resetRunStats = () => {
     DoubleMapper.runs = 0;