Fix cluster request forwarding

anticluster.js

@@ -332,9 +332,9 @@ class AntiCluster
     {
         if (path == 'kv_txn')
         {
-            return ((!data.compare || !data.compare.length) &&
-                (!data.success || !data.success.filter(f => f.request_put || f.requestPut || f.request_delete_range || f.requestDeleteRange).length) &&
-                (!data.failure || !data.failure.filter(f => f.request_put || f.requestPut || f.request_delete_range || f.requestDeleteRange).length));
+            return (data.compare && data.compare.length ||
+                data.success && data.success.filter(f => f.request_put || f.requestPut || f.request_delete_range || f.requestDeleteRange).length ||
+                data.failure && data.failure.filter(f => f.request_put || f.requestPut || f.request_delete_range || f.requestDeleteRange).length);
         }
         return path != 'kv_range';
     }

antietcd-app.js

@@ -12,10 +12,10 @@ License: Mozilla Public License 2.0 or Vitastor Network Public License 1.1
 
 Usage:
 
-${process.argv[0]} ${process.argv[1]} \
-    [--cert ssl.crt] [--key ssl.key] [--port 12379] \
-    [--data data.gz] [--persist-filter ./filter.js] [--persist_interval 500] \
-    [--node_id node1 --cluster_key abcdef --cluster node1=http://localhost:12379,node2=http://localhost:12380,node3=http://localhost:12381] \
+${process.argv[0]} ${process.argv[1]} \n\
+    [--cert ssl.crt] [--key ssl.key] [--port 12379] \n\
+    [--data data.gz] [--persist-filter ./filter.js] [--persist_interval 500] \n\
+    [--node_id node1 --cluster_key abcdef --cluster node1=http://localhost:12379,node2=http://localhost:12380,node3=http://localhost:12381] \n\
     [other options]
 
 Supported etcd REST APIs:
@@ -43,6 +43,10 @@ HTTP:
     Require TLS client certificates signed by <ca> or by default CA to connect.
 --ws_keepalive_interval 30000
     Client websocket ping (keepalive) interval in milliseconds
+--no_merge_watch off
+    Antietcd merges all watcher events into a single websocket message to provide
+    more ordering guarantees. Specify 1/on/true/yes for this option to disable this
+    behaviour.
 
 Persistence:
 

antietcd.js

@@ -31,6 +31,10 @@ class AntiEtcd extends EventEmitter
         this.cluster = null;
         this.stored_term = 0;
         this.cfg = cfg;
+        this.cfg.no_merge_watch = this.cfg.no_merge_watch == 1 ||
+            this.cfg.no_merge_watch == 'on' ||
+            this.cfg.no_merge_watch == 'yes' ||
+            this.cfg.no_merge_watch == 'true';
         this.loading = false;
         this.stopped = false;
         this.inflight = 0;
@@ -120,7 +124,7 @@ class AntiEtcd extends EventEmitter
             let done = 0;
             await new Promise((allOk, allNo) =>
             {
-                res.map(promise => promise.then(res =>
+                res.map(promise => promise.then(r =>
                 {
                     if ((++done) == res.length)
                         allOk();
@@ -411,9 +415,9 @@ class AntiEtcd extends EventEmitter
     }
 
     // public watch API
-    async create_watch(params, callback)
+    async create_watch(params, callback, stream_id)
     {
-        const watch = this.etctree.api_create_watch({ ...params, watch_id: null }, callback);
+        const watch = this.etctree.api_create_watch({ ...params, watch_id: null }, callback, stream_id);
         if (!watch.created)
         {
             throw new RequestError(400, 'Requested watch revision is compacted', { compact_revision: watch.compact_revision });
@@ -478,7 +482,7 @@ class AntiEtcd extends EventEmitter
         return this.etctree.api_keepalive_lease(data);
     }
 
-    _handle_maintenance_status(data)
+    _handle_maintenance_status(/*data*/)
     {
         const raft = this.cluster && this.cluster.raft;
         return {
@@ -516,8 +520,9 @@ class AntiEtcd extends EventEmitter
             const create_request = msg.create_request;
             if (!create_request.watch_id || !client.watches[create_request.watch_id])
             {
+                client.send_cb = client.send_cb || (msg => socket.send(JSON.stringify(msg)));
                 const watch = this.etctree.api_create_watch(
-                    { ...create_request, watch_id: null }, (msg) => socket.send(JSON.stringify(msg))
+                    { ...create_request, watch_id: null }, client.send_cb, (this.cfg.no_merge_watch ? null : 'C'+client_id)
                 );
                 if (!watch.created)
                 {

etctree.js

@@ -544,7 +544,7 @@ class EtcTree
         }
     }
 
-    api_create_watch(req, send)
+    api_create_watch(req, send, stream_id)
     {
         const { parts, all } = this._get_range(req);
         if (req.start_revision && this.compact_revision && this.compact_revision > req.start_revision)
@@ -566,6 +566,7 @@ class EtcTree
             this.watchers[watch_id] = {
                 paths: [],
                 send,
+                stream_id,
             };
         }
         this.watchers[watch_id].paths.push(parts);
@@ -667,15 +668,15 @@ class EtcTree
             {
                 if (this.watchers[wid])
                 {
-                    by_watcher[wid] = by_watcher[wid] || { header: { revision: this.mod_revision }, events: {} };
-                    by_watcher[wid].events[notif.key] = conv;
+                    const stream_id = this.watchers[wid].stream_id || wid;
+                    by_watcher[stream_id] = by_watcher[stream_id] || { send: this.watchers[wid].send, events: {} };
+                    by_watcher[stream_id].events[notif.key] = conv;
                 }
             }
         }
-        for (const wid in by_watcher)
+        for (const stream_id in by_watcher)
         {
-            by_watcher[wid].events = Object.values(by_watcher[wid].events);
-            this.watchers[wid].send({ result: by_watcher[wid] });
+            by_watcher[stream_id].send({ result: { header: { revision: this.mod_revision }, events: Object.values(by_watcher[stream_id].events) } });
         }
     }
 

etctree.spec.js

@@ -91,6 +91,38 @@ tests['watch'] = async () =>
     t.destroy();
 };
 
+tests['merge watch'] = async () =>
+{
+    const t = new EtcTree();
+    const sent = [];
+    const send = (event) => sent.push(event);
+    expect(
+        await t.api_txn({ success: [ { request_put: { key: '/vitastor//config/pgs', value: { items: {} } } } ] }),
+        { header: { revision: 1 }, succeeded: true, responses: [ { response_put: {} } ] }
+    );
+    expect(
+        t.api_create_watch({ watch_id: 1, key: '/vitastor/config/pgs' }, send, 'X1' /* stream_id */),
+        { header: { revision: 1 }, watch_id: 1, created: true }
+    );
+    expect(
+        t.api_create_watch({ watch_id: 2, key: '/vitastor/pg/history/', range_end: '/vitastor/pg/history0' }, send, 'X1' /* stream_id */),
+        { header: { revision: 1 }, watch_id: 2, created: true },
+    );
+    expect(sent, []);
+    expect(
+        await t.api_txn({ success: [
+            { request_put: { key: '/vitastor/config/pgs', value: { items: { 1: { 1: { osd_set: [ 1, 2, 3 ] } } } } } },
+            { request_put: { key: '/vitastor/pg/history/1/1', value: { all_peers: [ 1, 2, 3, 4, 5 ] } } },
+        ] }),
+        { header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} }, { response_put: {} } ] }
+    );
+    expect(sent, [ { result: { header: { revision: 2 }, events: [
+        { type: 'PUT', kv: { key: '/vitastor/config/pgs', value: { items: { 1: { 1: { osd_set: [ 1, 2, 3 ] } } } }, mod_revision: 2 } },
+        { type: 'PUT', kv: { key: '/vitastor/pg/history/1/1', value: { all_peers: [ 1, 2, 3, 4, 5 ] }, mod_revision: 2 } },
+    ] } } ]);
+    t.destroy();
+};
+
 tests['lease'] = async () =>
 {
     const t = new EtcTree();