Check concurrent modification in graph iterators (#367)

Summary:
A client of `Graph` is able to (e.g.) invoke `nodes()` to get a node
iterator, iterate over some of the nodes, then change the nodes by
adding or removing the nodes on the original graph. The semantics of
what to do here are not clear: ES6 specifies semantics for `Map` and
`Set`, but they have counterintuitive consequences: for instance, you
can get a `Set` iterator to yield the same value twice. Most collection
implementations in the Java standard library prohibit this entirely. In
this commit, we adopt the latter approach.

A caveat of this implementation is that a graph object may not be
mutated more than 2^53 − 1 (`Number.MAX_SAFE_INTEGER`) times. Clients
who need to mutate a graph more than nine quadrillion times are
encouraged to reconsider their data model.

Paired with @decentralion.

Test Plan:
Unit tests added. Run `yarn travis`.

wchargin-branch: comod-check

src/v3/core/graph.js

@@ -42,6 +42,8 @@ export type NeighborsOptions = {|
 
 export opaque type GraphJSON = any; // TODO
 
+type ModificationCount = number;
+
 export class Graph {
   // A node `n` is in the graph if `_nodes.has(n)`.
   //
@@ -55,16 +57,48 @@ export class Graph {
   _inEdges: Map<NodeAddressT, Edge[]>;
   _outEdges: Map<NodeAddressT, Edge[]>;
 
+  // Incremented each time that any change is made to the graph. Used to
+  // check for comodification.
+  _modificationCount: ModificationCount;
+
   constructor(): void {
+    this._modificationCount = 0;
     this._nodes = new Set();
     this._edges = new Map();
     this._inEdges = new Map();
     this._outEdges = new Map();
   }
 
+  _checkForComodification(since: ModificationCount) {
+    // TODO(perf): Consider eliding this in production.
+    const now = this._modificationCount;
+    if (now === since) {
+      return;
+    }
+    if (now > since) {
+      throw new Error("Concurrent modification detected");
+    }
+    if (now < since) {
+      throw new Error(
+        "Invariant violation: expected modification count in the future"
+      );
+    }
+  }
+
+  _markModification() {
+    // TODO(perf): Consider eliding this in production.
+    if (this._modificationCount >= Number.MAX_SAFE_INTEGER) {
+      throw new Error(
+        `Graph cannot be modified more than ${this._modificationCount} times.`
+      );
+    }
+    this._modificationCount++;
+  }
+
   addNode(a: NodeAddressT): this {
     NodeAddress.assertValid(a);
     this._nodes.add(a);
+    this._markModification();
     return this;
   }
 
@@ -79,6 +113,7 @@ export class Graph {
       }
     }
     this._nodes.delete(a);
+    this._markModification();
     return this;
   }
 
@@ -87,8 +122,18 @@ export class Graph {
     return this._nodes.has(a);
   }
 
-  *nodes(): Iterator<NodeAddressT> {
-    yield* this._nodes;
+  nodes(): Iterator<NodeAddressT> {
+    return this._nodesIterator(this._modificationCount);
+  }
+
+  *_nodesIterator(
+    initialModificationCount: ModificationCount
+  ): Iterator<NodeAddressT> {
+    for (const node of this._nodes) {
+      this._checkForComodification(initialModificationCount);
+      yield node;
+    }
+    this._checkForComodification(initialModificationCount);
   }
 
   addEdge(edge: Edge): this {
@@ -117,12 +162,14 @@ export class Graph {
       }
     }
     this._edges.set(edge.address, edge);
+    this._markModification();
     return this;
   }
 
   removeEdge(address: EdgeAddressT): this {
     EdgeAddress.assertValid(address);
     this._edges.delete(address);
+    this._markModification();
     return this;
   }
 
@@ -136,8 +183,16 @@ export class Graph {
     return this._edges.get(address);
   }
 
-  *edges(): Iterator<Edge> {
-    yield* this._edges.values();
+  edges(): Iterator<Edge> {
+    return this._edgesIterator(this._modificationCount);
+  }
+
+  *_edgesIterator(initialModificationCount: ModificationCount): Iterator<Edge> {
+    for (const edge of this._edges.values()) {
+      this._checkForComodification(initialModificationCount);
+      yield edge;
+    }
+    this._checkForComodification(initialModificationCount);
   }
 
   neighbors(node: NodeAddressT, options: NeighborsOptions): Iterator<Neighbor> {

src/v3/core/graph.test.js

@@ -43,6 +43,28 @@ describe("core/graph", () => {
       });
     }
 
+    it("throws when trying to perform an unsafe number of modifications", () => {
+      const g = new Graph();
+      g.addNode(NodeAddress.fromParts(["one"]));
+      g.addNode(NodeAddress.fromParts(["two"]));
+      // skip a few
+      g._modificationCount = Number.MAX_SAFE_INTEGER - 1;
+      g.addNode(NodeAddress.fromParts(["ninety-nine"]));
+      expect(() => {
+        g.addNode(NodeAddress.fromParts(["boom"]));
+      }).toThrow("cannot be modified");
+    });
+
+    it("throws in case of modification count reset", () => {
+      const g = new Graph();
+      g.addNode(NodeAddress.fromParts(["stop"]));
+      const iterator = g.nodes();
+      g._modificationCount--;
+      expect(() => {
+        iterator.next();
+      }).toThrow("modification count in the future");
+    });
+
     describe("node methods", () => {
       describe("error on", () => {
         const p = Graph.prototype;
@@ -81,6 +103,21 @@ describe("core/graph", () => {
             );
           });
         });
+
+        describe("concurrent modification in `nodes`", () => {
+          it("while in the middle of iteration", () => {
+            const g = new Graph().addNode(NodeAddress.fromParts(["node"]));
+            const iterator = g.nodes();
+            g._modificationCount++;
+            expect(() => iterator.next()).toThrow("Concurrent modification");
+          });
+          it("at exhaustion", () => {
+            const g = new Graph();
+            const iterator = g.nodes();
+            g._modificationCount++;
+            expect(() => iterator.next()).toThrow("Concurrent modification");
+          });
+        });
       });
 
       describe("work on", () => {
@@ -126,6 +163,35 @@ describe("core/graph", () => {
           expect(Array.from(graph.nodes()).sort()).toEqual([n2, n1]);
         });
       });
+
+      describe("change the modification count", () => {
+        it("on addNode, when a node is added", () => {
+          const g = new Graph();
+          const before = g._modificationCount;
+          g.addNode(NodeAddress.fromParts(["hello"]));
+          expect(g._modificationCount).not.toEqual(before);
+        });
+        it("on addNode, even when the node already exists", () => {
+          const node = NodeAddress.fromParts(["hello"]);
+          const g = new Graph().addNode(node);
+          const before = g._modificationCount;
+          g.addNode(node);
+          expect(g._modificationCount).not.toEqual(before);
+        });
+        it("on removeNode, when a node is removed", () => {
+          const node = NodeAddress.fromParts(["hello"]);
+          const g = new Graph().addNode(node);
+          const before = g._modificationCount;
+          g.removeNode(node);
+          expect(g._modificationCount).not.toEqual(before);
+        });
+        it("on removeNode, even when the node does not exist", () => {
+          const g = new Graph();
+          const before = g._modificationCount;
+          g.removeNode(NodeAddress.fromParts(["hello"]));
+          expect(g._modificationCount).not.toEqual(before);
+        });
+      });
     });
 
     describe("edge methods", () => {
@@ -225,6 +291,27 @@ describe("core/graph", () => {
             });
           });
         });
+
+        describe("concurrent modification in `edges`", () => {
+          it("while in the middle of iteration", () => {
+            const g = new Graph()
+              .addNode(NodeAddress.fromParts(["node"]))
+              .addEdge({
+                address: EdgeAddress.fromParts(["edge"]),
+                src: NodeAddress.fromParts(["node"]),
+                dst: NodeAddress.fromParts(["node"]),
+              });
+            const iterator = g.edges();
+            g._modificationCount++;
+            expect(() => iterator.next()).toThrow("Concurrent modification");
+          });
+          it("at exhaustion", () => {
+            const g = new Graph();
+            const iterator = g.edges();
+            g._modificationCount++;
+            expect(() => iterator.next()).toThrow("Concurrent modification");
+          });
+        });
       });
 
       describe("on a graph", () => {
@@ -335,6 +422,41 @@ describe("core/graph", () => {
         });
       });
     });
+
+    describe("change the modification count", () => {
+      const src = () => NodeAddress.fromParts(["fst"]);
+      const dst = () => NodeAddress.fromParts(["snd"]);
+      const edge = () => ({
+        address: EdgeAddress.fromParts(["bridge"]),
+        src: src(),
+        dst: dst(),
+      });
+      const baseGraph = () => new Graph().addNode(src()).addNode(dst());
+      it("on addEdge, when an edge is added", () => {
+        const g = baseGraph();
+        const before = g._modificationCount;
+        g.addEdge(edge());
+        expect(g._modificationCount).not.toEqual(before);
+      });
+      it("on addEdge, even when the edge already exists", () => {
+        const g = baseGraph().addEdge(edge());
+        const before = g._modificationCount;
+        g.addEdge(edge());
+        expect(g._modificationCount).not.toEqual(before);
+      });
+      it("on removeEdge, when an edge is removed", () => {
+        const g = baseGraph().addEdge(edge());
+        const before = g._modificationCount;
+        g.removeEdge(edge().address);
+        expect(g._modificationCount).not.toEqual(before);
+      });
+      it("on removeEdge, even when the edge does not exist", () => {
+        const g = new Graph();
+        const before = g._modificationCount;
+        g.removeEdge(edge().address);
+        expect(g._modificationCount).not.toEqual(before);
+      });
+    });
   });
 
   describe("edgeToString", () => {