Implement client

.eslintrc.js

@@ -0,0 +1,49 @@
+module.exports = {
+    "env": {
+        "es6": true,
+        "node": true
+    },
+    "extends": [
+        "eslint:recommended",
+        "plugin:node/recommended"
+    ],
+    "parserOptions": {
+        "ecmaVersion": 2020
+    },
+    "plugins": [
+    ],
+    "rules": {
+        "indent": [
+            "error",
+            4
+        ],
+        "brace-style": [
+            "error",
+            "allman",
+            { "allowSingleLine": true }
+        ],
+        "linebreak-style": [
+            "error",
+            "unix"
+        ],
+        "semi": [
+            "error",
+            "always"
+        ],
+        "no-useless-escape": [
+            "off"
+        ],
+        "no-control-regex": [
+            "off"
+        ],
+        "no-empty": [
+            "off"
+        ],
+        "no-process-exit": [
+            "off"
+        ],
+        "node/shebang": [
+            "off"
+        ]
+    }
+};

anticli.js

@@ -0,0 +1,224 @@
+#!/usr/bin/node
+
+const fsp = require('fs').promises;
+const http = require('http');
+const https = require('https');
+
+class AntiEtcdCli
+{
+    static parse(args)
+    {
+        const cmd = [];
+        const options = {};
+        for (let i = 2; i < args.length; i++)
+        {
+            const arg = args[i].toLowerCase().replace(/^--(.+)$/, (m, m1) => '--'+m1.replace(/-/g, '_'));
+            if (arg === '-h' || arg === '--help')
+            {
+                console.error(
+                    'USAGE:\n'+
+                    '  anticli.js [OPTIONS] put <key> [<value>]\n'+
+                    '  anticli.js [OPTIONS] get <key> [-p|--prefix] [-v|--print-value-only] [-k|--keys-only]\n'+
+                    '  anticli.js [OPTIONS] del <key> [-p|--prefix]\n'+
+                    'OPTIONS:\n'+
+                    '  [--endpoints|-e http://node1:2379,http://node2:2379,http://node3:2379]\n'+
+                    '  [--timeout 1000]'
+                );
+                process.exit();
+            }
+            else if (arg == '-e' || arg == '--endpoints')
+            {
+                options['endpoints'] = args[++i].split(/\s*[,\s]+\s*/);
+            }
+            else if (arg == '-p' || arg == '--prefix')
+            {
+                options['prefix'] = true;
+            }
+            else if (arg == '-v' || arg == '--print_value_only')
+            {
+                options['print_value_only'] = true;
+            }
+            else if (arg == '-k' || arg == '--keys_only')
+            {
+                options['keys_only'] = true;
+            }
+            else if (arg[0] == '-' && arg[1] !== '-')
+            {
+                console.error('Unknown option '+arg);
+                process.exit(1);
+            }
+            else if (arg.substr(0, 2) == '--')
+            {
+                options[arg.substr(2)] = args[++i];
+            }
+            else
+            {
+                cmd.push(arg);
+            }
+        }
+        if (!cmd.length || cmd[0] != 'get' && cmd[0] != 'put' && cmd[0] != 'del')
+        {
+            console.log('Supported commands: get, put, del. Use --help to see details');
+            process.exit(1);
+        }
+        return [ cmd, options ];
+    }
+
+    async run(cmd, options)
+    {
+        this.options = options;
+        if (!this.options.endpoints)
+        {
+            this.options.endpoints = [ 'http://localhost:2379' ];
+        }
+        if (cmd[0] == 'get')
+        {
+            await this.get(cmd.slice(1));
+        }
+        else if (cmd[0] == 'put')
+        {
+            await this.put(cmd[1], cmd.length > 2 ? cmd[2] : undefined);
+        }
+        else if (cmd[0] == 'del')
+        {
+            await this.del(cmd.slice(1));
+        }
+    }
+
+    async get(keys)
+    {
+        if (this.options.prefix)
+        {
+            keys = keys.map(k => k.replace(/\/+$/, ''));
+        }
+        const txn = { success: keys.map(key => ({ request_range: this.options.prefix ? { key: b64(key+'/'), range_end: b64(key+'0') } : { key: b64(key) } })) };
+        const res = await this.request('/v3/kv/txn', txn);
+        for (const r of (res||{}).responses||[])
+        {
+            if (r.response_range)
+            {
+                for (const kv of r.response_range.kvs)
+                {
+                    if (!this.options.print_value_only)
+                    {
+                        console.log(de64(kv.key));
+                    }
+                    if (!this.options.keys_only)
+                    {
+                        console.log(de64(kv.value));
+                    }
+                }
+            }
+        }
+    }
+
+    async put(key, value)
+    {
+        if (value === undefined)
+        {
+            value = await fsp.readFile(0, { encoding: 'utf-8' });
+        }
+        const res = await this.request('/v3/kv/put', { key: b64(key), value: b64(value) });
+        if (res && res.header)
+        {
+            console.log('OK');
+        }
+    }
+
+    async del(keys)
+    {
+        if (this.options.prefix)
+        {
+            keys = keys.map(k => k.replace(/\/+$/, ''));
+        }
+        const txn = { success: keys.map(key => ({ request_delete_range: this.options.prefix ? { key: b64(key+'/'), range_end: b64(key+'0') } : { key: b64(key) } })) };
+        const res = await this.request('/v3/kv/txn', txn);
+        for (const r of (res||{}).responses||[])
+        {
+            if (r.response_delete_range)
+            {
+                console.log(r.response_delete_range.deleted);
+            }
+        }
+    }
+
+    async request(path, body)
+    {
+        for (const url of this.options.endpoints)
+        {
+            const cur_url = url.replace(/\/+$/, '')+path;
+            try
+            {
+                return (await POST(cur_url, body, this.options.timeout||1000)).json;
+            }
+            catch (e)
+            {
+                console.error(cur_url+': '+e.message);
+            }
+        }
+        return null;
+    }
+}
+
+function POST(url, body, timeout)
+{
+    return new Promise(ok =>
+    {
+        const body_text = Buffer.from(JSON.stringify(body));
+        let timer_id = timeout > 0 ? setTimeout(() =>
+        {
+            if (req)
+                req.abort();
+            req = null;
+            ok({ error: 'timeout' });
+        }, timeout) : null;
+        let req = (url.substr(0, 6).toLowerCase() == 'https://' ? https : http).request(url, { method: 'POST', headers: {
+            'Content-Type': 'application/json',
+            'Content-Length': body_text.length,
+        } }, (res) =>
+        {
+            if (!req)
+            {
+                return;
+            }
+            clearTimeout(timer_id);
+            let res_body = '';
+            res.setEncoding('utf8');
+            res.on('error', (error) => ok({ error }));
+            res.on('data', chunk => { res_body += chunk; });
+            res.on('end', () =>
+            {
+                if (res.statusCode != 200)
+                {
+                    ok({ error: res_body, code: res.statusCode });
+                    return;
+                }
+                try
+                {
+                    res_body = JSON.parse(res_body);
+                    ok({ response: res, json: res_body });
+                }
+                catch (e)
+                {
+                    ok({ error: e, response: res, body: res_body });
+                }
+            });
+        });
+        req.on('error', (error) => ok({ error }));
+        req.on('close', () => ok({ error: new Error('Connection closed prematurely') }));
+        req.write(body_text);
+        req.end();
+    });
+}
+
+function b64(str)
+{
+    return Buffer.from(str).toString('base64');
+}
+
+function de64(str)
+{
+    return Buffer.from(str, 'base64').toString();
+}
+
+new AntiEtcdCli().run(...AntiEtcdCli.parse(process.argv)).catch(console.error);

anticluster.js

@@ -0,0 +1,460 @@
+const ws = require('ws');
+
+const TinyRaft = require('./tinyraft.js');
+const { runCallbacks, RequestError } = require('./common.js');
+
+const LEADER_MISMATCH = 'raft leader/term mismatch';
+
+class AntiCluster
+{
+    constructor(antietcd)
+    {
+        this.antietcd = antietcd;
+        this.cfg = antietcd.cfg;
+        this.cluster_connections = {};
+        this.last_request_id = 1;
+        this.subrequests = {};
+        this.synced = false;
+        this.wait_sync = [];
+        if (!this.cfg.node_id || !this.cfg.cluster_key)
+        {
+            throw new Error('node_id and cluster_key are required in configuration if cluster is set');
+        }
+        if (!(this.cfg.cluster instanceof Object))
+        {
+            this.cfg.cluster = (''+this.cfg.cluster).trim().split(/[\s,]*,[\s,]*/)
+                .reduce((a, c) => { c = c.split(/\s*=\s*/); a[c[0]] = c[1]; return a; }, {});
+        }
+        this.raft = new TinyRaft({
+            nodes: Object.keys(this.cfg.cluster),
+            nodeId: this.cfg.node_id,
+            heartbeatTimeout: this.cfg.heartbeat_timeout,
+            electionTimeout: this.cfg.election_timeout,
+            initialTerm: this.antietcd.stored_term,
+            send: (to, msg) => this._sendRaftMessage(to, msg),
+        });
+        this.raft.on('change', (event) => this._handleRaftChange(event));
+        this.raft.start();
+        // Connect to all nodes and reconnect forever
+        for (const node_id in this.cfg.cluster)
+        {
+            this.connectToNode(node_id);
+        }
+    }
+
+    connectToNode(node_id)
+    {
+        if (node_id != this.cfg.node_id && this.cfg.cluster[node_id] &&
+            (!this.cluster_connections[node_id] || !this.antietcd.clients[this.cluster_connections[node_id]]))
+        {
+            const socket = new ws.WebSocket(this.cfg.cluster[node_id].replace(/^http/, 'ws'));
+            const client_id = this.antietcd.startWebsocket(socket, () => setTimeout(() => this.connectToNode(node_id), this.cfg.reconnect_interval||1000));
+            this.cluster_connections[node_id] = client_id;
+            socket.on('open', () =>
+            {
+                if (this.antietcd.clients[client_id])
+                {
+                    this.antietcd.clients[client_id].ready = true;
+                    this.antietcd.clients[client_id].raft_node_id = node_id;
+                    this.antietcd.clients[client_id].addr = socket._socket.remoteAddress+':'+socket._socket.remotePort;
+                    socket.send(JSON.stringify({ identify: { key: this.cfg.cluster_key, node_id: this.cfg.node_id } }));
+                }
+            });
+        }
+    }
+
+    _peerRequest(client, request, timeout)
+    {
+        const request_id = this.last_request_id++;
+        request.request_id = request_id;
+        client.socket.send(JSON.stringify(request));
+        const req = this.subrequests[request_id] = { client_id: client.id };
+        const promise = new Promise(ok => req.cb = ok);
+        req.timer_id = setTimeout(() => this._completeRequest(null, request_id, { error: 'timeout' }), timeout);
+        return promise;
+    }
+
+    async replicateChange(msg)
+    {
+        if (this.raft.state !== TinyRaft.LEADER)
+        {
+            return;
+        }
+        for (const follower of this.raft.followers)
+        {
+            if (follower != this.cfg.node_id)
+            {
+                const client = this._getPeer(follower);
+                if (!client)
+                {
+                    // One of peers is unavailable - immediate failure, request should be retried
+                    console.log('Lost peer connection during replication - restarting election');
+                    this.raft.start();
+                    throw new Error('Replication failed');
+                }
+            }
+        }
+        const promises = [];
+        for (const follower of this.raft.followers)
+        {
+            if (follower != this.cfg.node_id)
+            {
+                const client = this._getPeer(follower);
+                const promise = this._peerRequest(client, { replicate: msg, term: this.raft.term }, this.cfg.replication_timeout||1000);
+                promises.push(promise);
+            }
+        }
+        const results = await Promise.all(promises);
+        for (const result of results)
+        {
+            if (!result)
+            {
+                // One of peers is unavailable - immediate failure, request should be retried
+                console.log('Lost peer connection during replication - restarting election');
+                this.raft.start();
+                throw new Error('Replication failed');
+            }
+        }
+    }
+
+    _completeRequest(client_id, request_id, result)
+    {
+        const req = this.subrequests[request_id];
+        if (!req || client_id && req.client_id != client_id)
+        {
+            return;
+        }
+        delete this.subrequests[request_id];
+        if (req.timer_id && result !== null)
+        {
+            clearTimeout(req.timer_id);
+        }
+        req.cb(result);
+    }
+
+    _handleRaftChange(event)
+    {
+        console.log(
+            'Raft '+this.cfg.node_id+': '+(event.state == TinyRaft.FOLLOWER ? 'following '+event.leader : event.state)+
+            ', term '+event.term+(event.state == TinyRaft.LEADER ? ', followers: '+event.followers.join(', ') : '')
+        );
+        if (event.state == TinyRaft.LEADER)
+        {
+            // (Re)sync with the new set of followers
+            this._resync(event.followers);
+            this.antietcd.etctree.resume_leases();
+        }
+        else
+        {
+            this.synced = false;
+            this.antietcd.etctree.pause_leases();
+        }
+    }
+
+    _resync(followers)
+    {
+        this.synced = false;
+        if (!this.resync_state)
+        {
+            this.resync_state = {
+                dumps: {},
+                loads: {},
+            };
+            this.resync_state.dumps[this.cfg.node_id] = { ...this.antietcd.etctree.dump(), term: this.antietcd.stored_term };
+        }
+        const seen = {};
+        for (const f of followers)
+        {
+            seen[f] = true;
+            if (f != this.cfg.node_id && !(f in this.resync_state.dumps))
+            {
+                const client = this._getPeer(f);
+                if (client)
+                {
+                    this.resync_state.dumps[f] = null;
+                    this._peerRequest(client, { request: {}, handler: 'dump' }, this.cfg.dump_timeout||5000).then(res =>
+                    {
+                        if (this.resync_state && client.raft_node_id &&
+                            (client.raft_node_id in this.resync_state.dumps))
+                        {
+                            if (res.error)
+                            {
+                                console.error(client.raft_node_id+' dump failed with error: '+res.error);
+                            }
+                            else
+                            {
+                                console.log('Got dump from '+client.raft_node_id+' with stored term '+res.term);
+                            }
+                            this.resync_state.dumps[client.raft_node_id] = res.error ? null : res;
+                            this._continueResync();
+                        }
+                    });
+                }
+            }
+        }
+        for (const f in this.resync_state.dumps)
+        {
+            if (!seen[f])
+            {
+                delete this.resync_state.dumps[f];
+            }
+        }
+        this._continueResync();
+    }
+
+    _continueResync()
+    {
+        let max_term = -1, with_max = [];
+        for (const follower in this.resync_state.dumps)
+        {
+            const dump = this.resync_state.dumps[follower];
+            if (!dump)
+            {
+                // Some dump(s) are still pending
+                return;
+            }
+            if (dump.term > max_term)
+            {
+                max_term = dump.term;
+                with_max = [ follower ];
+            }
+            else if (dump.term == max_term)
+            {
+                with_max.push(follower);
+            }
+        }
+        if (max_term < 0 || with_max.length == 0)
+        {
+            throw new Error('BUG: no max term during resync');
+        }
+        with_max = with_max.filter(w => w != this.cfg.node_id);
+        console.log('Local term '+this.antietcd.stored_term+', max follower term '+max_term+' at nodes '+with_max.join(', '));
+        // Merge databases of all nodes with maximum term
+        // Force other nodes to replicate the merged DB, throwing away their own states
+        for (let i = 0; i < with_max.length; i++)
+        {
+            const update_only = !(i == 0 && this.antietcd.stored_term != max_term);
+            console.log(update_only ? 'Updating database from node '+with_max[i]+' state' : 'Copying node '+with_max[i]+' state');
+            this.antietcd.etctree.load(this.resync_state.dumps[with_max[i]], update_only);
+        }
+        let wait = 0;
+        const load_request = { term: this.raft.term, load: this.antietcd.etctree.dump() };
+        for (const follower in this.resync_state.dumps)
+        {
+            if (follower != this.cfg.node_id)
+            {
+                const dump = this.resync_state.dumps[follower];
+                if (dump.term <= max_term)
+                {
+                    const client = this._getPeer(follower);
+                    if (!client)
+                    {
+                        console.log('Lost peer connection during resync - restarting election');
+                        this.raft.start();
+                        return;
+                    }
+                    console.log('Copying state to '+follower);
+                    const loadstate = this.resync_state.loads[follower] = {};
+                    wait++;
+                    this._peerRequest(client, load_request, this.cfg.load_timeout||5000).then(res =>
+                    {
+                        loadstate.result = res;
+                        this._finishResync();
+                    });
+                }
+            }
+        }
+        if (!wait)
+        {
+            this._finishResync();
+        }
+    }
+
+    _finishResync()
+    {
+        if (Object.values(this.resync_state.dumps).filter(d => !d).length > 0 ||
+            Object.values(this.resync_state.loads).filter(d => !d.result).length > 0)
+        {
+            return;
+        }
+        // All current peers have copied the database, we can proceed
+        this.antietcd.stored_term = this.raft.term;
+        this.synced = true;
+        runCallbacks(this, 'wait_sync', []);
+        console.log('Synchronized with followers, new term is '+this.raft.term);
+    }
+
+    _isWrite(path, data)
+    {
+        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 path != 'kv_range';
+    }
+
+    async checkRaftState(path, requestUrl, data)
+    {
+        if (!this.raft)
+        {
+            return null;
+        }
+        if (requestUrl.searchParams.get('leaderonly') &&
+            this.raft.state != TinyRaft.LEADER)
+        {
+            throw new RequestError(503, 'Not leader');
+        }
+        if (this.raft.state == TinyRaft.CANDIDATE)
+        {
+            throw new RequestError(503, 'Quorum not available');
+        }
+        else if (this.raft.state == TinyRaft.FOLLOWER &&
+            (!this.cfg.stale_read || this._isWrite(path, data)))
+        {
+            // Forward to leader
+            return await this._forwardToLeader(path, data);
+        }
+        else if (!this.synced)
+        {
+            // Wait for initial sync for read-only requests
+            await new Promise(ok => this.wait_sync.push(ok));
+        }
+        return null;
+    }
+
+    async _forwardToLeader(handler, data)
+    {
+        const client = this._getPeer(this.raft.leader);
+        if (!client)
+        {
+            throw new RequestError(503, 'Leader is unavailable');
+        }
+        return await this._peerRequest(client, { handler, request: data }, this.cfg.forward_timeout||1000);
+    }
+
+    handleWsMsg(client, msg)
+    {
+        if (msg.raft)
+        {
+            if (client.raft_node_id)
+            {
+                this.raft.onReceive(client.raft_node_id, msg.raft);
+            }
+        }
+        else if (msg.identify)
+        {
+            if (msg.identify.key === this.cfg.cluster_key &&
+                msg.identify.node_id != this.cfg.node_id)
+            {
+                client.raft_node_id = msg.identify.node_id;
+                console.log('Got a connection from '+client.raft_node_id);
+            }
+        }
+        else if (msg.load)
+        {
+            this._handleLoadMsg(client, msg);
+        }
+        else if (msg.replicate)
+        {
+            this._handleReplicateMsg(client, msg).catch(console.error);
+        }
+        else if (msg.request)
+        {
+            this._handleRequestMsg(client, msg).catch(console.error);
+        }
+        else if (msg.reply)
+        {
+            this._completeRequest(client.id, msg.request_id, msg.reply);
+        }
+    }
+
+    async _handleRequestMsg(client, msg)
+    {
+        const cb = this.antietcd['handle_'+msg.handler];
+        if (cb)
+        {
+            try
+            {
+                let res = cb.call(this.antietcd, msg.request);
+                if (res instanceof Promise)
+                {
+                    res = await res;
+                }
+                client.socket.send(JSON.stringify({ request_id: msg.request_id, reply: res }));
+            }
+            catch (e)
+            {
+                console.error(e);
+                client.socket.send(JSON.stringify({ request_id: msg.request_id, reply: { error: e.message } }));
+            }
+        }
+        else
+        {
+            client.socket.send(JSON.stringify({ request_id: msg.request_id, reply: { error: 'unknown handler' } }));
+        }
+    }
+
+    _handleLoadMsg(client, msg)
+    {
+        if (client.raft_node_id && this.raft.state == TinyRaft.FOLLOWER &&
+            this.raft.leader === client.raft_node_id && this.raft.term == msg.term)
+        {
+            this.antietcd.etctree.load(msg.load);
+            this.antietcd.stored_term = msg.term;
+            this.synced = true;
+            runCallbacks(this, 'wait_sync', []);
+            console.log('Synchronized with leader, new term is '+msg.term);
+            client.socket.send(JSON.stringify({ request_id: msg.request_id, reply: {} }));
+        }
+        else
+        {
+            client.socket.send(JSON.stringify({ request_id: msg.request_id, reply: { error: LEADER_MISMATCH } }));
+        }
+    }
+
+    async _handleReplicateMsg(client, msg)
+    {
+        if (client.raft_node_id && this.raft.state == TinyRaft.FOLLOWER &&
+            this.raft.leader === client.raft_node_id && this.raft.term == msg.term)
+        {
+            await this.antietcd.etctree.apply_replication(msg.replicate);
+            client.socket.send(JSON.stringify({ request_id: msg.request_id, reply: {} }));
+        }
+        else
+        {
+            client.socket.send(JSON.stringify({ request_id: msg.request_id, reply: { error: LEADER_MISMATCH } }));
+        }
+    }
+
+    _getPeer(to)
+    {
+        if (to == this.cfg.node_id)
+        {
+            throw new Error('BUG: attempt to get connection to self');
+        }
+        const client_id = this.cluster_connections[to];
+        if (!client_id)
+        {
+            return null;
+        }
+        const client = this.antietcd.clients[client_id];
+        if (!client || !client.ready)
+        {
+            return null;
+        }
+        return client;
+    }
+
+    _sendRaftMessage(to, msg)
+    {
+        const client = this._getPeer(to);
+        if (client)
+        {
+            client.socket.send(JSON.stringify({ raft: msg }));
+        }
+    }
+}
+
+module.exports = AntiCluster;

antietcd.js

@@ -1,4 +1,6 @@
-const fs = require('fs');
+#!/usr/bin/node
+
+const fsp = require('fs').promises;
 const { URL } = require('url');
 const http = require('http');
 const https = require('https');
@@ -6,53 +8,101 @@ const https = require('https');
 const ws = require('ws');
 
 const EtcTree = require('./etctree.js');
-
-class RequestError
-{
-    constructor(code, text)
-    {
-        this.code = code;
-        this.message = text;
-    }
-}
+const AntiPersistence = require('./antipersistence.js');
+const AntiCluster = require('./anticluster.js');
+const { runCallbacks, RequestError } = require('./common.js');
 
 class AntiEtcd
 {
     constructor(cfg)
-    {
-        this.cfg = cfg;
-    }
-
-    run()
     {
         this.clients = {};
         this.client_id = 1;
         this.etctree = new EtcTree(true);
+        this.persistence = null;
+        this.cluster = null;
+        this.stored_term = 0;
+        this.cfg = cfg;
+        this.loading = false;
+        this.stopped = false;
+        this.inflight = 0;
+        this.wait_inflight = [];
+    }
+
+    async run()
+    {
+        if (this.cfg.data || this.cfg.cluster)
+        {
+            this.etctree.set_replicate_watcher(msg => this.persistAndReplicate(msg));
+        }
+        if (this.cfg.data)
+        {
+            this.persistence = new AntiPersistence(this);
+            // Load data from disk
+            await this.persistence.load();
+            // Set exit hook
+            const on_stop_cb = () => this.onStop();
+            process.on('SIGINT', on_stop_cb);
+            process.on('SIGTERM', on_stop_cb);
+            process.on('SIGQUIT', on_stop_cb);
+        }
+        if (this.cfg.cluster)
+        {
+            this.cluster = new AntiCluster(this);
+        }
         if (this.cfg.cert)
         {
-            this.tls = { key: fs.readFileSync(this.cfg.key), cert: fs.readFileSync(this.cfg.cert) };
+            this.tls = { key: await fsp.readFile(this.cfg.key), cert: await fsp.readFile(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));
+        // eslint-disable-next-line no-unused-vars
+        this.wss.on('connection', (conn, req) => this.startWebsocket(conn, null));
         this.server.listen(this.cfg.port || 2379);
     }
 
-    fail(res, code, text)
+    async onStop()
     {
-        res.writeHead(code);
-        res.write(text);
-        res.end();
+        if (this.stopped)
+        {
+            return;
+        }
+        this.stopped = true;
+        // Wait until all requests complete
+        while (this.inflight > 0)
+        {
+            await new Promise(ok => this.wait_inflight.push(ok));
+        }
+        if (this.persistence)
+        {
+            await this.persistence.persist();
+        }
+        process.exit(0);
+    }
+
+    async persistAndReplicate(msg)
+    {
+        if (this.persistence)
+        {
+            await this.persistence.persistChange(msg);
+        }
+        if (this.cluster)
+        {
+            await this.cluster.replicateChange(msg);
+        }
     }
 
     handleRequest(req, res)
     {
         let data = [];
         req.on('data', (chunk) => data.push(chunk));
-        req.on('end', () =>
+        req.on('end', async () =>
         {
+            this.inflight++;
             data = Buffer.concat(data);
             let body = '';
             let code = 200;
@@ -76,7 +126,7 @@ class AntiEtcd
                 {
                     throw new RequestError(400, 'body should be JSON object');
                 }
-                reply = this.runHandler(req, data, res);
+                reply = await this.runHandler(req, data, res);
                 reply = JSON.stringify(reply);
             }
             catch (e)
@@ -93,34 +143,47 @@ class AntiEtcd
                     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();
+            try
+            {
+                // Access log
+                console.log(
+                    new Date().toISOString()+
+                    ' '+(req.headers['x-forwarded-for'] || (req.socket.remoteAddress + ':' + req.socket.remotePort))+
+                    ' '+req.method+' '+req.url+' '+code+'\n  '+body.replace(/\n/g, '\\n')+
+                    '\n  '+reply.replace(/\n/g, '\\n')
+                );
+                reply = Buffer.from(reply);
+                res.writeHead(200, {
+                    'Content-Type': 'application/json',
+                    'Content-Length': reply.length,
+                });
+                res.write(reply);
+                res.end();
+            }
+            catch (e)
+            {
+                console.error(e);
+            }
+            this.inflight--;
+            if (!this.inflight)
+            {
+                runCallbacks(this, 'wait_inflight', []);
+            }
         });
     }
 
-    startWebsocket(socket, req)
+    startWebsocket(socket, reconnect)
     {
         const client_id = this.client_id++;
         this.clients[client_id] = {
+            id: client_id,
+            addr: socket._socket ? socket._socket.remoteAddress+':'+socket._socket.remotePort : '',
             socket,
             alive: true,
             watches: {},
         };
         socket.on('pong', () => this.clients[client_id].alive = true);
-        socket.on('error', console.error);
+        socket.on('error', e => console.error(e.syscall === 'connect' ? e.message : e));
         const pinger = setInterval(() =>
         {
             if (!this.clients[client_id])
@@ -160,10 +223,16 @@ class AntiEtcd
             clearInterval(pinger);
             delete this.clients[client_id];
             socket.terminate();
+            if (reconnect)
+            {
+                reconnect();
+            }
         });
+        return client_id;
     }
 
-    runHandler(req, data, res)
+    // eslint-disable-next-line no-unused-vars
+    async runHandler(req, data, res)
     {
         // v3/kv/txn
         // v3/kv/range
@@ -173,10 +242,22 @@ class AntiEtcd
         // v3/lease/keepalive
         // v3/lease/revoke O_o
         // v3/kv/lease/revoke O_o
+        if (this.stopped)
+        {
+            throw new RequestError(502, 'Server is stopping');
+        }
         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, '_');
+            if (this.cluster)
+            {
+                const res = await this.cluster.checkRaftState(path, requestUrl, data);
+                if (res)
+                {
+                    return res;
+                }
+            }
             const cb = this['handle_'+path];
             if (cb)
             {
@@ -184,33 +265,42 @@ class AntiEtcd
                 {
                     throw new RequestError(405, 'Please use POST method');
                 }
-                return cb.call(this, data);
+                const res = cb.call(this, data);
+                if (res instanceof Promise)
+                {
+                    return await res;
+                }
+                return res;
             }
         }
+        else if (requestUrl.pathname == '/dump')
+        {
+            return this.handle_dump(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)
+    async handle_kv_txn(data)
     {
-        return this.etctree.api_txn(data);
+        return await this.etctree.api_txn(data);
     }
 
-    handle_kv_range(data)
+    async handle_kv_range(data)
     {
-        const r = this.etctree.api_txn({ success: [ { request_range: data } ] });
+        const r = await this.etctree.api_txn({ success: [ { request_range: data } ] });
         return { header: r.header, ...r.responses[0].response_range };
     }
 
-    handle_kv_put(data)
+    async handle_kv_put(data)
     {
-        const r = this.etctree.api_txn({ success: [ { request_put: data } ] });
+        const r = await this.etctree.api_txn({ success: [ { request_put: data } ] });
         return { header: r.header, ...r.responses[0].response_put };
     }
 
-    handle_kv_deleterange(data)
+    async handle_kv_deleterange(data)
     {
-        const r = this.etctree.api_txn({ success: [ { request_delete_range: data } ] });
+        const r = await this.etctree.api_txn({ success: [ { request_delete_range: data } ] });
         return { header: r.header, ...r.responses[0].response_delete_range };
     }
 
@@ -234,13 +324,20 @@ class AntiEtcd
         return this.etctree.api_keepalive_lease(data);
     }
 
+    // eslint-disable-next-line no-unused-vars
+    handle_dump(data)
+    {
+        return { ...this.etctree.dump(), term: this.stored_term };
+    }
+
     handleMessage(client_id, msg, socket)
     {
+        const client = this.clients[client_id];
+        console.log(new Date().toISOString()+' '+client.addr+' '+(client.raft_node_id || '-')+' -> '+JSON.stringify(msg));
         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])
+            if (!create_request.watch_id || !client.watches[create_request.watch_id])
             {
                 const watch = this.etctree.api_create_watch(
                     { ...create_request, watch_id: null }, (msg) => socket.send(JSON.stringify(msg))
@@ -251,19 +348,19 @@ class AntiEtcd
                 }
                 else
                 {
-                    create_request.watch_id ||= watch.watch_id;
-                    this.clients[client_id].watches[create_request.watch_id] = watch.watch_id;
+                    create_request.watch_id = create_request.watch_id || watch.watch_id;
+                    client.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];
+            const mapped_id = client.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];
+                delete client.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 } }));
             }
         }
@@ -271,6 +368,14 @@ class AntiEtcd
         {
             socket.send(JSON.stringify({ result: { header: { revision: this.etctree.mod_revision } } }));
         }
+        else
+        {
+            if (!this.cluster)
+            {
+                return;
+            }
+            this.cluster.handleWsMsg(client, msg);
+        }
     }
 
     unsubscribeClient(client_id)
@@ -287,4 +392,81 @@ class AntiEtcd
     }
 }
 
-new AntiEtcd({ port: 12379 }).run();
+function vitastor_persist_filter(prefix)
+{
+    return (key, value) =>
+    {
+        if (key.substr(0, prefix.length+'/osd/stats/'.length) == prefix+'/osd/stats/')
+        {
+            if (value)
+            {
+                try
+                {
+                    value = JSON.parse(value);
+                    value = JSON.stringify({
+                        bitmap_granularity: value.bitmap_granularity || undefined,
+                        data_block_size: value.data_block_size || undefined,
+                        host: value.host || undefined,
+                        immediate_commit: value.immediate_commit || undefined,
+                    });
+                }
+                catch (e)
+                {
+                    console.error('invalid JSON in '+key+' = '+value+': '+e);
+                    value = {};
+                }
+            }
+            else
+            {
+                value = undefined;
+            }
+            return value;
+        }
+        else if (key.substr(0, prefix.length+'/osd/'.length) == prefix+'/osd/' ||
+            key.substr(0, prefix.length+'/inode/stats/'.length) == prefix+'/inode/stats/' ||
+            key.substr(0, prefix.length+'/pg/stats/'.length) == prefix+'/pg/stats/' ||
+            key.substr(0, prefix.length+'/pool/stats/'.length) == prefix+'/pool/stats/' ||
+            key == prefix+'/stats')
+        {
+            return undefined;
+        }
+        return value;
+    };
+}
+
+function parse()
+{
+    const options = {
+        persist_filter: vitastor_persist_filter('/vitastor'),
+    };
+    for (let i = 2; i < process.argv.length; i++)
+    {
+        const arg = process.argv[i].toLowerCase().replace(/^--(.+)$/, (m, m1) => '--'+m1.replace(/-/g, '_'));
+        if (arg === '-h' || arg === '--help')
+        {
+            console.error(
+                'USAGE:\n  '+process.argv[0]+' '+process.argv[1]+' [OPTIONS]\n'+
+                'OPTIONS:\n'+
+                '  [--cert ssl.crt] [--key ssl.key] [--port 12379]\n'+
+                '  [--data data.gz] [--vitastor-persist-filter /vitastor] [--no-persist-filter] [--persist_interval 500]\n'+
+                '  [--node_id node1 --cluster_key abcdef --cluster node1=http://localhost:12379,node2=http://localhost:12380,node3=http://localhost:12381]'
+            );
+            process.exit();
+        }
+        else if (arg == '--no_persist_filter')
+        {
+            options['persist_filter'] = null;
+        }
+        else if (arg == '--vitastor_persist_filter')
+        {
+            options['persist_filter'] = vitastor_persist_filter(process.argv[++i]||'');
+        }
+        else if (arg.substr(0, 2) == '--' && arg != '--persist_filter')
+        {
+            options[arg.substr(2)] = process.argv[++i];
+        }
+    }
+    return options;
+}
+
+new AntiEtcd(parse()).run().catch(console.error);

antipersistence.js

@@ -0,0 +1,134 @@
+const fs = require('fs');
+const fsp = require('fs').promises;
+const zlib = require('zlib');
+
+const EtcTree = require('./etctree.js');
+const { de64, runCallbacks } = require('./common.js');
+
+class AntiPersistence
+{
+    constructor(antietcd)
+    {
+        this.cfg = antietcd.cfg;
+        this.antietcd = antietcd;
+        this.prev_value = {};
+        this.persist_timer = null;
+        this.wait_persist = null;
+    }
+
+    async load()
+    {
+        // eslint-disable-next-line no-unused-vars
+        const [ err, stat ] = await new Promise(ok => fs.stat(this.cfg.data, (err, stat) => ok([ err, stat ])));
+        if (!err)
+        {
+            let data = await fsp.readFile(this.cfg.data);
+            data = await new Promise((ok, no) => zlib.gunzip(data, (err, res) => err ? no(err) : ok(res)));
+            data = JSON.parse(data);
+            this.loading = true;
+            this.antietcd.etctree.load(data);
+            this.loading = false;
+            this.antietcd.stored_term = data['term'] || 0;
+        }
+        else if (err.code != 'ENOENT')
+        {
+            throw err;
+        }
+    }
+
+    async persistChange(msg)
+    {
+        if (this.loading)
+        {
+            return;
+        }
+        if (!msg.events || !msg.events.length)
+        {
+            // lease-only changes don't need to be persisted
+            return;
+        }
+        if (this.cfg.persist_filter)
+        {
+            let changed = false;
+            for (const ev of msg.events)
+            {
+                if (ev.kv.lease)
+                {
+                    // Values with lease are never persisted
+                    const key = de64(ev.kv.key);
+                    if (this.prev_value[key] !== undefined)
+                    {
+                        delete this.prev_value[key];
+                        changed = true;
+                    }
+                }
+                else
+                {
+                    const key = de64(ev.kv.key);
+                    const filtered = this.cfg.persist_filter(key, ev.type === 'DELETE' ? undefined : de64(ev.kv.value));
+                    if (!EtcTree.eq(filtered, this.prev_value[key]))
+                    {
+                        this.prev_value[key] = filtered;
+                        changed = true;
+                    }
+                }
+                changed = true;
+            }
+            if (!changed)
+            {
+                return;
+            }
+        }
+        await this.schedulePersist();
+    }
+
+    async schedulePersist()
+    {
+        if (!this.cfg.persist_interval)
+        {
+            await this.persist();
+            return;
+        }
+        if (!this.persist_timer)
+        {
+            this.persist_timer = setTimeout(() =>
+            {
+                this.persist_timer = null;
+                this.persist().catch(console.error);
+            }, this.cfg.persist_interval);
+        }
+    }
+
+    async persist()
+    {
+        if (!this.cfg.data)
+        {
+            return;
+        }
+        while (this.wait_persist)
+        {
+            await new Promise(ok => this.wait_persist.push(ok));
+        }
+        this.wait_persist = [];
+        try
+        {
+            let dump = this.antietcd.etctree.dump(true);
+            dump['term'] = this.antietcd.stored_term;
+            dump = JSON.stringify(dump);
+            dump = await new Promise((ok, no) => zlib.gzip(dump, (err, res) => err ? no(err) : ok(res)));
+            const fh = await fsp.open(this.cfg.data+'.tmp', 'w');
+            await fh.writeFile(dump);
+            await fh.sync();
+            await fh.close();
+            await fsp.rename(this.cfg.data+'.tmp', this.cfg.data);
+        }
+        catch (e)
+        {
+            console.error(e);
+            process.exit(1);
+        }
+        runCallbacks(this, 'wait_persist', null);
+    }
+}
+
+module.exports = AntiPersistence;

common.js

@@ -0,0 +1,34 @@
+class RequestError
+{
+    constructor(code, text)
+    {
+        this.code = code;
+        this.message = text;
+    }
+}
+
+function de64(k)
+{
+    if (k == null) // null or undefined
+        return k;
+    return Buffer.from(k, 'base64').toString();
+}
+
+function runCallbacks(obj, key, new_value)
+{
+    const cbs = obj[key];
+    obj[key] = new_value;
+    if (cbs)
+    {
+        for (const cb of cbs)
+        {
+            cb();
+        }
+    }
+}
+
+module.exports = {
+    RequestError,
+    de64,
+    runCallbacks,
+};

package.json

@@ -0,0 +1,29 @@
+{
+  "name": "tinyraft",
+  "version": "1.0.0",
+  "description": "Tiny & abstract Raft leader election algorithm",
+  "main": "tinyraft.js",
+  "scripts": {
+    "lint": "eslint common.js anticli.js antipersistence.js anticluster.js antietcd.js etctree.js etctree.spec.js tinyraft.js tinyraft.spec.js",
+    "test": "node etctree.spec.js && node tinyraft.spec.js"
+  },
+  "repository": {
+    "type": "git",
+    "url": "https://git.yourcmc.ru/vitalif/tinyraft"
+  },
+  "keywords": [
+    "raft"
+  ],
+  "author": "Vitaliy Filippov",
+  "license": "MPL-2.0",
+  "engines": {
+    "node": ">=12.0.0"
+  },
+  "devDependencies": {
+    "eslint": "^8.0.0",
+    "eslint-plugin-node": "^11.1.0"
+  },
+  "dependencies": {
+    "ws": "^8.17.0"
+  }
+}