Add Anti-Etcd - etcd mock, already sufficient to run Vitastor tests

antietcd.js

@@ -0,0 +1,288 @@
+const fs = require('fs');
+const { URL } = require('url');
+const http = require('http');
+const https = require('https');
+
+const ws = require('ws');
+
+const EtcTree = require('./etctree.js');
+
+class RequestError
+{
+    constructor(code, text)
+    {
+        this.code = code;
+        this.message = text;
+    }
+}
+
+class AntiEtcd
+{
+    constructor(cfg)
+    {
+        this.cfg = cfg;
+    }
+
+    run()
+    {
+        this.clients = {};
+        this.client_id = 1;
+        this.etctree = new EtcTree(true);
+        if (this.cfg.cert)
+        {
+            this.tls = { key: fs.readFileSync(this.cfg.key), cert: fs.readFileSync(this.cfg.cert) };
+            this.server = https.createServer(this.tls, (req, res) => this.handleRequest(req, res));
+        }
+        else
+            this.server = http.createServer((req, res) => this.handleRequest(req, res));
+        this.wss = new ws.WebSocketServer({ server: this.server });
+        this.wss.on('connection', (conn, req) => this.startWebsocket(conn, req));
+        this.server.listen(this.cfg.port || 2379);
+    }
+
+    fail(res, code, text)
+    {
+        res.writeHead(code);
+        res.write(text);
+        res.end();
+    }
+
+    handleRequest(req, res)
+    {
+        let data = [];
+        req.on('data', (chunk) => data.push(chunk));
+        req.on('end', () =>
+        {
+            data = Buffer.concat(data);
+            let body = '';
+            let code = 200;
+            let reply;
+            try
+            {
+                if (req.headers['content-type'] != 'application/json')
+                {
+                    throw new RequestError(400, 'content-type should be application/json');
+                }
+                body = data.toString();
+                try
+                {
+                    data = data.length ? JSON.parse(data) : {};
+                }
+                catch (e)
+                {
+                    throw new RequestError(400, 'body should be valid JSON');
+                }
+                if (!(data instanceof Object) || data instanceof Array)
+                {
+                    throw new RequestError(400, 'body should be JSON object');
+                }
+                reply = this.runHandler(req, data, res);
+                reply = JSON.stringify(reply);
+            }
+            catch (e)
+            {
+                if (e instanceof RequestError)
+                {
+                    code = e.code;
+                    reply = e.message;
+                }
+                else
+                {
+                    console.error(e);
+                    code = 500;
+                    reply = 'Internal error: '+e.message;
+                }
+            }
+            // Access log
+            console.log(
+                new Date().toISOString()+
+                ' '+(req.headers['x-forwarded-for'] || req.socket.remoteAddress)+
+                ' '+req.method+' '+req.url+' '+code+'\n  '+body.replace(/\n/g, '\\n')+
+                '\n  '+reply.replace(/\n/g, '\\n')
+            );
+            // FIXME: Access log :req[X-Forwarded-For] [:date[clf]] pid=:pid ":method :url HTTP/:http-version" :status :res[content-length] ":referrer" ":user-agent" - :response-time ms
+            reply = Buffer.from(reply);
+            res.writeHead(200, {
+                'Content-Type': 'application/json',
+                'Content-Length': reply.length,
+            });
+            res.write(reply);
+            res.end();
+        });
+    }
+
+    startWebsocket(socket, req)
+    {
+        const client_id = this.client_id++;
+        this.clients[client_id] = {
+            socket,
+            alive: true,
+            watches: {},
+        };
+        socket.on('pong', () => this.clients[client_id].alive = true);
+        socket.on('error', console.error);
+        const pinger = setInterval(() =>
+        {
+            if (!this.clients[client_id])
+            {
+                return;
+            }
+            if (!this.clients[client_id].alive)
+            {
+                return socket.terminate();
+            }
+            this.clients[client_id].alive = false;
+            socket.ping(() => {});
+        }, 30000);
+        socket.on('message', (msg) =>
+        {
+            try
+            {
+                msg = JSON.parse(msg);
+            }
+            catch (e)
+            {
+                socket.send(JSON.stringify({ error: 'bad-json' }));
+                return;
+            }
+            if (!msg)
+            {
+                socket.send(JSON.stringify({ error: 'empty-message' }));
+            }
+            else
+            {
+                this.handleMessage(client_id, msg, socket);
+            }
+        });
+        socket.on('close', () =>
+        {
+            this.unsubscribeClient(client_id);
+            clearInterval(pinger);
+            delete this.clients[client_id];
+            socket.terminate();
+        });
+    }
+
+    runHandler(req, data, res)
+    {
+        // v3/kv/txn
+        // v3/kv/range
+        // v3/kv/put
+        // v3/kv/deleterange
+        // v3/lease/grant
+        // v3/lease/keepalive
+        // v3/lease/revoke O_o
+        // v3/kv/lease/revoke O_o
+        const requestUrl = new URL(req.url, 'http://'+(req.headers.host || 'localhost'));
+        if (requestUrl.pathname.substr(0, 4) == '/v3/')
+        {
+            const path = requestUrl.pathname.substr(4).replace(/\/+$/, '').replace(/\/+/g, '_');
+            const cb = this['handle_'+path];
+            if (cb)
+            {
+                if (req.method != 'POST')
+                {
+                    throw new RequestError(405, 'Please use POST method');
+                }
+                return cb.call(this, data);
+            }
+        }
+        throw new RequestError(404, 'Supported APIs: /v3/kv/txn, /v3/kv/range, /v3/kv/put, /v3/kv/deleterange, '+
+            '/v3/lease/grant, /v3/lease/revoke, /v3/kv/lease/revoke, /v3/lease/keepalive');
+    }
+
+    handle_kv_txn(data)
+    {
+        return this.etctree.api_txn(data);
+    }
+
+    handle_kv_range(data)
+    {
+        const r = this.etctree.api_txn({ success: [ { request_range: data } ] });
+        return { header: r.header, ...r.responses[0].response_range };
+    }
+
+    handle_kv_put(data)
+    {
+        const r = this.etctree.api_txn({ success: [ { request_put: data } ] });
+        return { header: r.header, ...r.responses[0].response_put };
+    }
+
+    handle_kv_deleterange(data)
+    {
+        const r = this.etctree.api_txn({ success: [ { request_delete_range: data } ] });
+        return { header: r.header, ...r.responses[0].response_delete_range };
+    }
+
+    handle_lease_grant(data)
+    {
+        return this.etctree.api_grant_lease(data);
+    }
+
+    handle_lease_revoke(data)
+    {
+        return this.etctree.api_revoke_lease(data);
+    }
+
+    handle_kv_lease_revoke(data)
+    {
+        return this.etctree.api_revoke_lease(data);
+    }
+
+    handle_lease_keepalive(data)
+    {
+        return this.etctree.api_keepalive_lease(data);
+    }
+
+    handleMessage(client_id, msg, socket)
+    {
+        if (msg.create_request)
+        {
+            // FIXME progress_notify, filters, prev_kv
+            const create_request = msg.create_request;
+            if (!create_request.watch_id || !this.clients[client_id].watches[create_request.watch_id])
+            {
+                const watch = this.etctree.api_create_watch(
+                    { ...create_request, watch_id: null }, (msg) => socket.send(JSON.stringify(msg))
+                );
+                if (!watch.created)
+                {
+                    socket.send(JSON.stringify({ result: { header: { revision: this.etctree.mod_revision }, watch_id: create_request.watch_id, ...watch } }));
+                }
+                else
+                {
+                    create_request.watch_id ||= watch.watch_id;
+                    this.clients[client_id].watches[create_request.watch_id] = watch.watch_id;
+                    socket.send(JSON.stringify({ result: { header: { revision: this.etctree.mod_revision }, watch_id: create_request.watch_id, created: true } }));
+                }
+            }
+        }
+        else if (msg.cancel_request)
+        {
+            const mapped_id = this.clients[client_id].watches[msg.cancel_request.watch_id];
+            if (mapped_id)
+            {
+                this.etctree.api_cancel_watch({ watch_id: mapped_id });
+                delete this.clients[client_id].watches[msg.cancel_request.watch_id];
+                socket.send(JSON.stringify({ result: { header: { revision: this.etctree.mod_revision }, watch_id: msg.cancel_request.watch_id, canceled: true } }));
+            }
+        }
+        else if (msg.progress_request)
+        {
+            socket.send(JSON.stringify({ result: { header: { revision: this.etctree.mod_revision } } }));
+        }
+    }
+
+    unsubscribeClient(client_id)
+    {
+        if (!this.clients[client_id])
+            return;
+        for (const watch_id in this.clients[client_id].watches)
+        {
+            const mapped_id = this.clients[client_id].watches[watch_id];
+            this.etctree.api_cancel_watch({ watch_id: mapped_id });
+        }
+    }
+}
+
+new AntiEtcd({ port: 12379 }).run();

etctree.js

@@ -1,4 +1,5 @@
 const crypto = require('crypto');
+const stableStringify = require('./stable-stringify.js');
 
 /*type TreeNode = {
     value?: any,
@@ -13,38 +14,53 @@ const crypto = require('crypto');
 
 class EtcTree
 {
-    constructor()
+    constructor(use_base64)
     {
         this.state = {};
         this.leases = {};
         this.watchers = {};
         this.watcher_id = 0;
         this.mod_revision = 0;
+        this.use_base64 = use_base64;
+    }
+
+    de64(k)
+    {
+        if (k == null) // null or undefined
+            return k;
+        return this.use_base64 ? Buffer.from(k, 'base64').toString() : k;
+    }
+
+    b64(k)
+    {
+        if (k == null) // null or undefined
+            return k;
+        return this.use_base64 ? Buffer.from(k).toString('base64') : k;
     }
 
     check(chk)
     {
-        const parts = this.key_parts(chk.key);
+        const parts = this.key_parts(this.de64(chk.key));
         const { cur } = this.get_subtree(parts, false, false);
         let check_value, ref_value;
         if (chk.target === 'MOD')
         {
-            check_value = cur.mod_revision || 0;
+            check_value = cur && cur.mod_revision || 0;
             ref_value = chk.mod_revision || 0;
         }
         else if (chk.target === 'CREATE')
         {
-            check_value = cur.create_revision || 0;
+            check_value = cur && cur.create_revision || 0;
             ref_value = chk.create_revision || 0;
         }
         else if (chk.target === 'VERSION')
         {
-            check_value = cur.version || 0;
+            check_value = cur && cur.version || 0;
             ref_value = chk.version || 0;
         }
         else if (chk.target === 'LEASE')
         {
-            check_value = cur.lease;
+            check_value = cur && cur.lease;
             ref_value = chk.lease;
         }
         else
@@ -64,14 +80,14 @@ class EtcTree
 
     key_parts(key)
     {
-        const parts = key.replace(/\/\/+/g, '/').replace(/^\/|\/$/g, '');
+        const parts = key.replace(/\/\/+/g, '/').replace(/\/$/g, ''); // trim beginning?
         return parts === '' ? [] : parts.split('/');
     }
 
     get_range(req)
     {
-        const key = req.key;
+        const key = this.de64(req.key);
-        const end = req.range_end;
+        const end = this.de64(req.range_end);
         if (end != null && (key[key.length-1] != '/' || end[end.length-1] != '0' ||
             end.substr(0, end.length-1) !== key.substr(0, key.length-1)))
         {
@@ -125,11 +141,11 @@ class EtcTree
         let id;
         while (!id || this.leases[id])
         {
-            id = crypto.randomBytes(8).toString('base64');
+            id = crypto.randomBytes(8).toString('hex');
         }
         const timer_id = setTimeout(() => this.api_revoke_lease({ ID: id }), req.TTL*1000);
         this.leases[id] = { ttl: req.TTL, timer_id, keys: {} };
-        return { ID: id };
+        return { header: { revision: this.mod_revision }, ID: id, TTL: req.TTL };
     }
 
     api_keepalive_lease(req)
@@ -140,9 +156,10 @@ class EtcTree
             throw new Error('unknown lease');
         }
         clearTimeout(this.leases[id].timer_id);
-        const ttl = this.leases[id].TTL;
+        const ttl = this.leases[id].ttl;
         this.leases[id].timer_id = setTimeout(() => this.api_revoke_lease({ ID: id }), ttl*1000);
-        return { TTL: ttl };
+        // extra wrapping in { result: ... }
+        return { result: { header: { revision: this.mod_revision }, ID: id, TTL: ''+ttl } };
     }
 
     api_revoke_lease(req)
@@ -158,14 +175,16 @@ class EtcTree
             this.txn_action({ request_delete_range: { key } }, next_revision, notifications);
         }
         this.notify(notifications);
+        return { header: { revision: this.mod_revision } };
     }
 
     api_create_watch(req, send)
     {
         const { parts, all } = this.get_range(req);
-        if (req.start_revision && req.start_revision < this.mod_revision)
+        if (req.start_revision && this.compact_revision && this.compact_revision > req.start_revision)
         {
-            throw new Error('history storage is not implemented');
+            // Deletions up to this.compact_revision are forgotten
+            return { compact_revision: this.compact_revision };
         }
         let watch_id = req.watch_id;
         if (watch_id instanceof Object)
@@ -195,9 +214,41 @@ class EtcTree
             cur.key_watchers = cur.key_watchers || [];
             cur.key_watchers.push(watch_id);
         }
+        if (req.start_revision && req.start_revision < this.mod_revision)
+        {
+            // Send initial changes
+            setImmediate(() =>
+            {
+                const events = [];
+                const { cur } = this.get_subtree([], false, false);
+                this.get_modified(events, cur, null, req.start_revision);
+                send({ result: { header: { revision: this.mod_revision }, events } });
+            });
+        }
         return { watch_id, created: true };
     }
 
+    get_modified(events, cur, prefix, min_rev)
+    {
+        if (cur.mod_revision >= min_rev)
+        {
+            events.push({
+                type: cur.value == null ? 'DELETE' : 'PUT',
+                kv: cur.value == null ? { key: this.b64(prefix === null ? '' : prefix) } : {
+                    key: this.b64(prefix),
+                    value: this.b64(cur.value),
+                },
+            });
+        }
+        if (cur.children)
+        {
+            for (const k in cur.children)
+            {
+                this.get_modified(events, cur.children[k], prefix === null ? k : prefix+'/'+k, min_rev);
+            }
+        }
+    }
+
     api_cancel_watch(watch_id)
     {
         if (this.watchers[watch_id])
@@ -224,6 +275,7 @@ class EtcTree
             }
             delete this.watchers[watch_id];
         }
+        return { canceled: true };
     }
 
     api_txn({ compare, success, failure })
@@ -237,7 +289,7 @@ class EtcTree
             responses.push(this.txn_action(req, next_revision, notifications));
         }
         this.notify(notifications);
-        return { revision: this.mod_revision, succeeded: !failed, responses };
+        return { header: { revision: this.mod_revision }, succeeded: !failed, responses };
     }
 
     notify(notifications)
@@ -251,58 +303,61 @@ class EtcTree
         {
             const watchers = notif.watchers;
             delete notif.watchers;
+            const conv = { type: ('value' in notif) ? 'PUT' : 'DELETE', kv: notif };
             for (const wid of watchers)
             {
                 if (this.watchers[wid])
                 {
                     by_watcher[wid] = by_watcher[wid] || { header: { revision: this.mod_revision }, events: {} };
-                    by_watcher[wid].events[notif.key] = notif;
+                    by_watcher[wid].events[notif.key] = conv;
                 }
             }
         }
         for (const wid in by_watcher)
         {
             by_watcher[wid].events = Object.values(by_watcher[wid].events);
-            this.watchers[wid].send(by_watcher[wid]);
+            this.watchers[wid].send({ result: by_watcher[wid] });
         }
     }
 
     txn_action(req, cur_revision, notifications)
     {
-        if (req.request_range)
+        if (req.request_range || req.requestRange)
         {
-            // FIXME: limit, revision(-), sort_order, sort_target, serializable(-), keys_only,
+            const request_range = req.request_range || req.requestRange;
+            // FIXME: limit, revision(-), sort_order, sort_target, serializable(-),
             // count_only, min_mod_revision, max_mod_revision, min_create_revision, max_create_revision
-            const { parts, all } = this.get_range(req.request_range);
+            const { parts, all } = this.get_range(request_range);
             const { cur } = this.get_subtree(parts, false, false);
             const kvs = [];
             if (cur)
             {
-                this.get_all(kvs, cur, all, parts.join('/'), req.request_range);
+                this.get_all(kvs, cur, all, parts.join('/') || null, request_range);
             }
-            return { kvs };
+            return { response_range: { kvs } };
         }
-        else if (req.request_put)
+        else if (req.request_put || req.requestPut)
         {
+            const request_put = req.request_put || req.requestPut;
             // FIXME: prev_kv, ignore_value(?), ignore_lease(?)
-            const parts = this.key_parts(req.request_put.key);
+            const parts = this.key_parts(this.de64(request_put.key));
             const key = parts.join('/');
-            const value = req.request_put.value;
+            const value = this.de64(request_put.value);
             const { cur, watchers } = this.get_subtree(parts, true, true);
-            if (!eq(cur.value, value) || cur.lease != req.request_put.lease)
+            if (!eq(cur.value, value) || cur.lease != request_put.lease)
             {
                 if (cur.lease && this.leases[cur.lease])
                 {
                     delete this.leases[cur.lease].keys[key];
                 }
-                if (req.request_put.lease)
+                if (request_put.lease)
                 {
-                    if (!this.leases[req.request_put.lease])
+                    if (!this.leases[request_put.lease])
                     {
-                        throw new Error('unknown lease: '+req.request_put.lease);
+                        throw new Error('unknown lease: '+request_put.lease);
                     }
-                    cur.lease = req.request_put.lease;
+                    cur.lease = request_put.lease;
-                    this.leases[req.request_put.lease].keys[key] = true;
+                    this.leases[request_put.lease].keys[key] = true;
                 }
                 this.mod_revision = cur_revision;
                 cur.version = (cur.version||0) + 1;
@@ -312,25 +367,29 @@ class EtcTree
                     cur.create_revision = cur_revision;
                 }
                 cur.value = value;
-                const notify = { watchers, key, value, mod_revision: cur.mod_revision };
+                const notify = { watchers, key: this.b64(key), value: this.b64(value), mod_revision: cur.mod_revision };
                 if (cur.lease)
                 {
                     notify.lease = cur.lease;
                 }
                 notifications.push(notify);
             }
-            return {};
+            return { response_put: {} };
         }
-        else if (req.request_delete_range)
+        else if (req.request_delete_range || req.requestDeleteRange)
         {
+            const request_delete_range = req.request_delete_range || req.requestDeleteRange;
             // FIXME: prev_kv
-            const { parts, all } = this.get_range(req.request_delete_range);
+            const { parts, all } = this.get_range(request_delete_range);
             const { cur, watchers } = this.get_subtree(parts, false, true);
+            const prevcount = notifications.length;
             if (cur)
             {
-                this.delete_all(notifications, watchers, cur, all, parts.join('/'), cur_revision);
+                this.delete_all(notifications, watchers, cur, all, parts.join('/') || null, cur_revision);
             }
+            return { response_delete_range: { deleted: notifications.length-prevcount } };
         }
+        return {};
     }
 
     get_all(kvs, cur, all, prefix, req)
@@ -343,18 +402,18 @@ class EtcTree
         {
             if (req.keys_only)
             {
-                kvs.push({ key: prefix, mod_revision: cur.mod_revision });
+                kvs.push({ key: this.b64(prefix === null ? '' : prefix), mod_revision: cur.mod_revision });
             }
             else
             {
-                kvs.push({ key: prefix, value: cur.value, mod_revision: cur.mod_revision });
+                kvs.push({ key: this.b64(prefix === null ? '' : prefix), value: this.b64(cur.value), mod_revision: cur.mod_revision });
             }
         }
         if (all && cur.children)
         {
             for (let k in cur.children)
             {
-                this.get_all(kvs, cur.children[k], true, prefix === '' ? k : prefix+'/'+k, req);
+                this.get_all(kvs, cur.children[k], true, prefix === null ? k : prefix+'/'+k, req);
             }
         }
     }
@@ -364,18 +423,20 @@ class EtcTree
         if (cur.value != null)
         {
             // Do not actually forget the key until the deletion is confirmed by all replicas
+            // ...and until it's not required by watchers
             cur.value = null;
+            cur.version = 0;
             cur.create_revision = null;
             cur.mod_revision = cur_revision;
             this.mod_revision = cur_revision;
-            notifications.push({ watchers, key: prefix, mod_revision: cur_revision });
+            notifications.push({ watchers, key: this.b64(prefix === null ? '' : prefix), mod_revision: cur_revision });
         }
         if (all && cur.children)
         {
             for (let k in cur.children)
             {
                 const subw = cur.children[k].watchers ? [ ...watchers, ...cur.children[k].watchers ] : watchers;
-                this.delete_all(notifications, subw, cur.children[k], true, prefix === '' ? k : prefix+'/'+k, cur_revision);
+                this.delete_all(notifications, subw, cur.children[k], true, prefix === null ? k : prefix+'/'+k, cur_revision);
             }
         }
     }
@@ -385,9 +446,11 @@ function eq(a, b)
 {
     if (a instanceof Object || b instanceof Object)
     {
-        return JSON.stringify(a) === JSON.stringify(b);
+        return stableStringify(a) === stableStringify(b);
     }
     return a == b;
 }
 
+EtcTree.eq = eq;
+
 module.exports = EtcTree;

etctree.spec.js

@@ -1,59 +1,109 @@
 const EtcTree = require('./etctree.js');
 
-describe('EtcTree', () =>
+const tests = {};
+
+let cur_test = '';
+
+const expect = (a, b) =>
 {
-    it('should return the inserted value', () =>
+    if (!EtcTree.eq(a, b))
     {
-        const t = new EtcTree();
+        process.stderr.write(cur_test+' test:\nexpected: '+JSON.stringify(b)+'\nreal:     '+JSON.stringify(a)+'\n'+new Error().stack.replace(/^.*\n.*\n/, '')+'\n');
-        expect(t.api_txn({ success: [ { request_put: { key: '/vitastor//config/global', value: { hello: 'world' } } } ] }))
+        process.exit(1);
-            .toEqual({ succeeded: true, revision: 1, responses: [ {} ] });
+    }
-        expect(t.api_txn({ success: [ { request_range: { key: '/vitastor/config/global' } } ] }))
+};
-            .toEqual({ succeeded: true, revision: 1, responses: [ { kvs: [ { key: 'vitastor/config/global', mod_revision: 1, value: { hello: 'world' } } ] } ] });
+
-        expect(t.api_txn({ success: [ { request_range: { key: '/vitastor/config/', range_end: '/vitastor/config0' } } ] }))
+tests['read/write'] = () =>
-            .toEqual({ succeeded: true, revision: 1, responses: [ { kvs: [ { key: 'vitastor/config/global', mod_revision: 1, value: { hello: 'world' } } ] } ] });
+{
-        expect(t.api_txn({ success: [ { request_range: { key: '/vitasto/', range_end: '/vitasto0' } } ] }))
+    const t = new EtcTree();
-            .toEqual({ succeeded: true, revision: 1, responses: [ { kvs: [] } ] });
+    expect(
-        expect(t.api_txn({
+        t.api_txn({ success: [ { request_put: { key: '/vitastor//config/global', value: { hello: 'world' } } } ] }),
+        { header: { revision: 1 }, succeeded: true, responses: [ { response_put: {} } ] }
+    );
+    expect(
+        t.api_txn({ success: [ { request_range: { key: '/vitastor/config/global' } } ] }),
+        { header: { revision: 1 }, succeeded: true, responses: [ { response_range: {
+            kvs: [ { key: '/vitastor/config/global', mod_revision: 1, value: { hello: 'world' } } ],
+        } } ] }
+    );
+    expect(
+        t.api_txn({ success: [ { request_range: { key: '/vitastor/config/', range_end: '/vitastor/config0' } } ] }),
+        { header: { revision: 1 }, succeeded: true, responses: [ { response_range: {
+            kvs: [ { key: '/vitastor/config/global', mod_revision: 1, value: { hello: 'world' } } ],
+        } } ] }
+    );
+    expect(
+        t.api_txn({ success: [ { request_range: { key: '/vitasto/', range_end: '/vitasto0' } } ] }),
+        { header: { revision: 1 }, succeeded: true, responses: [ { response_range: { kvs: [] } } ] }
+    );
+    expect(
+        t.api_txn({
             compare: [ { key: '/vitastor/config/global', target: 'MOD', mod_revision: 1, result: 'LESS' } ],
             success: [ { request_put: { key: '/vitastor//config/global', value: { hello: 'world' } } } ],
-            failure: [ { request_range: { key: 'vitastor/config/global' } } ],
+            failure: [ { request_range: { key: '/vitastor/config/global' } } ],
-        })).toEqual({ succeeded: false, revision: 1, responses: [ { kvs: [ { key: 'vitastor/config/global', mod_revision: 1, value: { hello: 'world' } } ] } ] });
+        }),
-        expect(t.api_txn({
+        { header: { revision: 1 }, succeeded: false, responses: [ { response_range: {
+            kvs: [ { key: '/vitastor/config/global', mod_revision: 1, value: { hello: 'world' } } ],
+        } } ] }
+    );
+    expect(
+        t.api_txn({
             compare: [ { key: '/vitastor/config/global', target: 'MOD', mod_revision: 2, result: 'LESS' } ],
             success: [ { request_put: { key: '/vitastor//config/global', value: { hello: 'world2' } } } ]
-        })).toEqual({ succeeded: true, revision: 2, responses: [ {} ] });
+        }),
-        expect(t.api_txn({ success: [ { request_range: { key: '/vitastor/config/', range_end: '/vitastor/config0' } } ] }))
+        { header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} } ] }
-            .toEqual({ succeeded: true, revision: 2, responses: [ { kvs: [ { key: 'vitastor/config/global', mod_revision: 2, value: { hello: 'world2' } } ] } ] });
+    );
-    });
+    expect(
+        t.api_txn({ success: [ { request_range: { key: '/vitastor/config/', range_end: '/vitastor/config0' } } ] }),
+        { header: { revision: 2 }, succeeded: true, responses: [ { response_range: {
+            kvs: [ { key: '/vitastor/config/global', mod_revision: 2, value: { hello: 'world2' } } ],
+        } } ] }
+    );
+};
 
-    it('should watch', () =>
+tests['watch'] = () =>
-    {
+{
-        const t = new EtcTree();
+    const t = new EtcTree();
-        const sent = [];
+    const sent = [];
-        const send = (event) => sent.push(event);
+    const send = (event) => sent.push(event);
-        expect(t.api_txn({ success: [ { request_put: { key: '/vitastor//config/global', value: { hello: 'world' } } } ] }))
+    expect(
-            .toEqual({ succeeded: true, revision: 1, responses: [ {} ] });
+        t.api_txn({ success: [ { request_put: { key: '/vitastor//config/global', value: { hello: 'world' } } } ] }),
-        expect(t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send))
+        { header: { revision: 1 }, succeeded: true, responses: [ { response_put: {} } ] }
-            .toEqual({ watch_id: 1, created: true });
+    );
-        expect(sent).toEqual([]);
+    expect(
-        expect(t.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', value: { ip: '1.2.3.4' } } } ] }))
+        t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send),
-            .toEqual({ succeeded: true, revision: 2, responses: [ {} ] });
+        { watch_id: 1, created: true }
-        expect(sent).toEqual([ { header: { revision: 2 }, events: [ { key: 'vitastor/osd/state/1', mod_revision: 2, value: { ip: '1.2.3.4' } } ] } ]);
+    );
-    });
+    expect(sent, []);
+    expect(
+        t.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', value: { ip: '1.2.3.4' } } } ] }),
+        { header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} } ] }
+    );
+    expect(sent, [ { result: { header: { revision: 2 }, events: [ { type: 'PUT', kv: { key: '/vitastor/osd/state/1', value: { ip: '1.2.3.4' }, mod_revision: 2 } } ] } } ]);
+};
 
-    it('should lease', async () =>
+tests['lease'] = async () =>
-    {
+{
-        const t = new EtcTree();
+    const t = new EtcTree();
-        const sent = [];
+    const sent = [];
-        const send = (event) => sent.push(event);
+    const send = (event) => sent.push(event);
-        const leaseID = t.api_grant_lease({ TTL: 0.5 }).ID;
+    const leaseID = t.api_grant_lease({ TTL: 0.5 }).ID;
-        expect(leaseID).not.toBeNull();
+    expect(leaseID != null, true);
-        expect(t.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.4' } } } ] }))
+    expect(
-            .toEqual({ succeeded: true, revision: 1, responses: [ {} ] });
+        t.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.4' } } } ] }),
-        expect(t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send))
+        { header: { revision: 1 }, succeeded: true, responses: [ { response_put: {} } ] }
-            .toEqual({ watch_id: 1, created: true });
+    );
-        expect(sent).toEqual([]);
+    expect(
-        await new Promise(ok => setTimeout(ok, 600));
+        t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send),
-        expect(sent).toEqual([ { header: { revision: 2 }, events: [ { key: 'vitastor/osd/state/1', mod_revision: 2 } ] } ]);
+        { watch_id: 1, created: true }
-    });
+    );
-});
+    expect(sent, []);
+    await new Promise(ok => setTimeout(ok, 600));
+    expect(sent, [ { result: { header: { revision: 2 }, events: [ { type: 'DELETE', kv: { key: '/vitastor/osd/state/1', mod_revision: 2 } } ] } } ]);
+};
+
+for (cur_test in tests)
+{
+    tests[cur_test]();
+    console.log(cur_test+' test: OK');
+}

stable-stringify.js

@@ -0,0 +1,78 @@
+// Copyright (c) Vitaliy Filippov, 2019+
+// License: MIT
+
+function stableStringify(obj, opts)
+{
+    if (!opts)
+        opts = {};
+    if (typeof opts === 'function')
+        opts = { cmp: opts };
+    let space = opts.space || '';
+    if (typeof space === 'number')
+        space = Array(space+1).join(' ');
+    const cycles = (typeof opts.cycles === 'boolean') ? opts.cycles : false;
+    const cmp = opts.cmp && (function (f)
+    {
+        return function (node)
+        {
+            return function (a, b)
+            {
+                let aobj = { key: a, value: node[a] };
+                let bobj = { key: b, value: node[b] };
+                return f(aobj, bobj);
+            };
+        };
+    })(opts.cmp);
+    const seen = new Map();
+    return (function stringify (parent, key, node, level)
+    {
+        const indent = space ? ('\n' + new Array(level + 1).join(space)) : '';
+        const colonSeparator = space ? ': ' : ':';
+        if (node === undefined)
+        {
+            return;
+        }
+        if (typeof node !== 'object' || node === null)
+        {
+            return JSON.stringify(node);
+        }
+        if (node instanceof Array)
+        {
+            const out = [];
+            for (let i = 0; i < node.length; i++)
+            {
+                const item = stringify(node, i, node[i], level+1) || JSON.stringify(null);
+                out.push(indent + space + item);
+            }
+            return '[' + out.join(',') + indent + ']';
+        }
+        else
+        {
+            if (seen.has(node))
+            {
+                if (cycles)
+                    return JSON.stringify('__cycle__');
+                throw new TypeError('Converting circular structure to JSON');
+            }
+            else
+                seen.set(node, true);
+            const keys = Object.keys(node).sort(cmp && cmp(node));
+            const out = [];
+            for (let i = 0; i < keys.length; i++)
+            {
+                const key = keys[i];
+                const value = stringify(node, key, node[key], level+1);
+                if (!value)
+                    continue;
+                const keyValue = JSON.stringify(key)
+                    + colonSeparator
+                    + value;
+                out.push(indent + space + keyValue);
+            }
+            seen.delete(node);
+            return '{' + out.join(',') + indent + '}';
+        }
+    })({ '': obj }, '', obj, 0);
+}
+
+module.exports = stableStringify;