Include start_revision when watching

README.md

@@ -27,6 +27,7 @@ Simplistic miniature etcd replacement based on [TinyRaft](https://git.yourcmc.ru
   - [/v3/lease/grant](#v3-lease-grant)
   - [/v3/lease/keepalive](#v3-lease-keepalive)
   - [/v3/lease/revoke or /v3/kv/lease/revoke](#v3-lease-revoke-or-v3-kv-lease-revoke)
+  - [/v3/maintenance/status](#v3-maintenance-status)
   - [Websocket-based watch APIs](#websocket-based-watch-apis)
 - [HTTP Error Codes](#http-error-codes)
 
@@ -48,8 +49,9 @@ Antietcd doesn't background itself, so use systemd or start-stop-daemon to run i
 
 ```
 node_modules/.bin/anticli [OPTIONS] put <key> [<value>]
-node_modules/.bin/anticli [OPTIONS] get <key> [-p|--prefix] [-v|--print-value-only] [-k|--keys-only]
+node_modules/.bin/anticli [OPTIONS] get <key> [-p|--prefix] [-v|--print-value-only] [-k|--keys-only] [--no-temp]
 node_modules/.bin/anticli [OPTIONS] del <key> [-p|--prefix]
+node_modules/.bin/anticli [OPTIONS] load [--with-lease] < dump.json
 ```
 
 For `put`, if `<value>` is not specified, it will be read from STDIN.
@@ -70,6 +72,9 @@ Options:
 <dt>--timeout 1000</dt>
 <dd>Specify request timeout in milliseconds</dd>
 
+<dt>--json or --write-out=json</dt>
+<dd>Print raw response in JSON</dd>
+
 </dl>
 
 ## Options
@@ -81,6 +86,9 @@ Options:
 <dt>--port 2379</dt>
 <dd>Listen port</dd>
 
+<dt>--ip 192.168.0.10</dt>
+<dd>Listen address (by default listen on all interfaces)</dd>
+
 <dt>--cert &lt;cert&gt;</dt>
 <dd>Use TLS with this certificate file (PEM format)</dd>
 
@@ -97,6 +105,9 @@ Specify &lt;ca&gt; = &lt;cert&gt; if your certificate is self-signed.</dd>
 <dt>--ws_keepalive_interval 30000</dt>
 <dd>Client websocket ping (keepalive) interval in milliseconds</dd>
 
+<dt>--use_base64 1</dt>
+<dd>Use base64 encoding of keys and values, like in etcd (enabled by default).</dd>
+
 </dl>
 
 ### Persistence
@@ -443,6 +454,33 @@ type LeaseRevokeResponse = {
 }
 ```
 
+### /v3/maintenance/status
+
+Request:
+
+```{}```
+
+Response:
+
+```ts
+type MaintenanceStatusResponse = {
+  header: {
+    member_id?: string,
+    revision: number,
+    compact_revision: number,
+    raft_term?: number,
+  },
+  version: string,
+  cluster?: { [string]: string },
+  leader?: string,
+  followers?: string[],
+  raftTerm?: string,
+  raftState?: 'leader'|'follower'|'candidate',
+  // dbSize actually reports process memory usage
+  dbSize: number,
+}
+```
+
 ### Websocket-based watch APIs
 
 Client-to-server message format:

anticli.js

@@ -4,6 +4,7 @@
 // (c) Vitaliy Filippov, 2024
 // License: Mozilla Public License 2.0 or Vitastor Network Public License 1.1
 
+const fs = require('fs');
 const fsp = require('fs').promises;
 const http = require('http');
 const https = require('https');
@@ -15,13 +16,14 @@ License: Mozilla Public License 2.0 or Vitastor Network Public License 1.1
 Usage:
 
 anticli.js [OPTIONS] put <key> [<value>]
-anticli.js [OPTIONS] get <key> [-p|--prefix] [-v|--print-value-only] [-k|--keys-only]
+anticli.js [OPTIONS] get <key> [-p|--prefix] [-v|--print-value-only] [-k|--keys-only] [--no-temp]
 anticli.js [OPTIONS] del <key> [-p|--prefix]
+anticli.js [OPTIONS] load [--with-lease] < dump.json
 
 Options:
 
 [--endpoints|-e http://node1:2379,http://node2:2379,http://node3:2379]
-[--cert cert.pem] [--key key.pem] [--timeout 1000]
+[--cert cert.pem] [--key key.pem] [--timeout 1000] [--json]
 `;
 
 class AntiEtcdCli
@@ -54,6 +56,19 @@ class AntiEtcdCli
             {
                 options['keys_only'] = true;
             }
+            else if (arg == '--with_lease')
+            {
+                options['with_lease'] = true;
+            }
+            else if (arg == '--write_out' && args[i+1] == 'json')
+            {
+                i++;
+                options['json'] = true;
+            }
+            else if (arg == '--json' || arg == '--write_out=json')
+            {
+                options['json'] = true;
+            }
             else if (arg[0] == '-' && arg[1] !== '-')
             {
                 process.stderr.write('Unknown option '+arg);
@@ -68,9 +83,9 @@ class AntiEtcdCli
                 cmd.push(arg);
             }
         }
-        if (!cmd.length || cmd[0] != 'get' && cmd[0] != 'put' && cmd[0] != 'del')
+        if (!cmd.length || cmd[0] != 'get' && cmd[0] != 'put' && cmd[0] != 'del' && cmd[0] != 'load')
         {
-            process.stderr.write('Supported commands: get, put, del. Use --help to see details\n');
+            process.stderr.write('Supported commands: get, put, del, load. Use --help to see details\n');
             process.exit(1);
         }
         return [ cmd, options ];
@@ -102,12 +117,51 @@ class AntiEtcdCli
         {
             await this.del(cmd.slice(1));
         }
+        else if (cmd[0] == 'load')
+        {
+            await this.load();
+        }
         // wait until output is fully flushed
         await new Promise(ok => process.stdout.write('', ok));
         await new Promise(ok => process.stderr.write('', ok));
         process.exit(0);
     }
 
+    async load()
+    {
+        const dump = JSON.parse(await new Promise((ok, no) => fs.readFile(0, { encoding: 'utf-8' }, (err, res) => err ? no(err) : ok(res))));
+        if (!dump.responses && !dump.kvs)
+        {
+            console.error('dump should be /kv/txn or /kv/range response in json format');
+            process.exit(1);
+        }
+        const success = [];
+        for (const r of (dump.responses
+            ? dump.responses.map(r => r.response_range).filter(r => r)
+            : [ dump ]))
+        {
+            for (const kv of r.kvs)
+            {
+                if (kv.value == null)
+                {
+                    console.error('dump should contain values');
+                    process.exit(1);
+                }
+                success.push({ request_put: { key: kv.key, value: kv.value, lease: this.options.with_lease ? kv.lease||undefined : undefined } });
+            }
+        }
+        const res = await this.request('/v3/kv/txn', { success });
+        if (this.options.json)
+        {
+            process.stdout.write(JSON.stringify(res));
+            return;
+        }
+        if (res.succeeded)
+        {
+            process.stdout.write('OK, loaded '+success.length+' values\n');
+        }
+    }
+
     async get(keys)
     {
         if (this.options.prefix)
@@ -116,6 +170,22 @@ class AntiEtcdCli
         }
         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);
+        if (this.options.notemp)
+        {
+            // Skip temporary values (values with lease)
+            for (const r of res.responses||[])
+            {
+                if (r.response_range)
+                {
+                    r.response_range.kvs = r.response_range.kvs.filter(kv => !kv.lease);
+                }
+            }
+        }
+        if (this.options.json)
+        {
+            process.stdout.write(JSON.stringify(keys.length == 1 ? res.responses[0].response_range : res));
+            return;
+        }
         for (const r of res.responses||[])
         {
             if (r.response_range)
@@ -139,7 +209,7 @@ class AntiEtcdCli
     {
         if (value === undefined)
         {
-            value = await fsp.readFile(0, { encoding: 'utf-8' });
+            value = await new Promise((ok, no) => fs.readFile(0, { encoding: 'utf-8' }, (err, res) => err ? no(err) : ok(res)));
         }
         const res = await this.request('/v3/kv/put', { key: b64(key), value: b64(value) });
         if (res.header)
@@ -175,18 +245,18 @@ class AntiEtcdCli
             {
                 if (res.json.error)
                 {
-                    process.stderr.write(cur_url+': '+res.json.error);
+                    process.stderr.write(cur_url+': '+res.json.error+'\n');
                     process.exit(1);
                 }
                 return res.json;
             }
             if (res.body)
             {
-                process.stderr.write(cur_url+': '+res.body);
+                process.stderr.write(cur_url+': '+res.body+'\n');
             }
             if (res.error)
             {
-                process.stderr.write(cur_url+': '+res.error);
+                process.stderr.write(cur_url+': '+res.error+'\n');
                 if (!res.response || !res.response.statusCode)
                 {
                     // This URL is unavailable

anticluster.js

@@ -226,7 +226,10 @@ class AntiCluster
                             }
                             else
                             {
-                                this._log('Got dump from '+client.raft_node_id+' with stored term '+res.term);
+                                this._log(
+                                    'Got dump from '+client.raft_node_id+' with stored term '+res.term+
+                                    ', mod_revision '+res.mod_revision+', compact_revision '+res.compact_revision
+                                );
                             }
                             this.resync_state.dumps[client.raft_node_id] = res.error ? null : res;
                             this._continueResync();
@@ -325,16 +328,19 @@ class AntiCluster
         this.antietcd.stored_term = this.raft.term;
         this.synced = true;
         runCallbacks(this, 'wait_sync', []);
-        this._log('Synchronized with followers, new term is '+this.raft.term);
+        this._log(
+            'Synchronized with followers, new term is '+this.raft.term+
+            ', mod_revision '+this.antietcd.etctree.mod_revision+', compact_revision '+this.antietcd.etctree.compact_revision
+        );
     }
 
     _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 (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';
     }
@@ -452,7 +458,10 @@ class AntiCluster
             this.antietcd.stored_term = msg.term;
             this.synced = true;
             runCallbacks(this, 'wait_sync', []);
-            this._log('Synchronized with leader, new term is '+msg.term);
+            this._log(
+                'Synchronized with leader, new term is '+this.raft.term+
+                ', mod_revision '+this.antietcd.etctree.mod_revision+', compact_revision '+this.antietcd.etctree.compact_revision
+            );
             client.socket.send(JSON.stringify({ request_id: msg.request_id, reply: {} }));
         }
         else

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:
@@ -30,6 +30,8 @@ HTTP:
 
 --port 2379
     Listen port
+--ip 192.168.0.10
+    Listen address (by default listen on all interfaces)
 --cert <cert>
     Use TLS with this certificate file (PEM format)
 --key <key>
@@ -41,6 +43,11 @@ 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
+--merge_watches 1
+    Antietcd merges all watcher events into a single websocket message to provide
+    more ordering/transaction guarantees. Set to 0 to disable this behaviour.
+--use_base64 1
+    Use base64 encoding of keys and values, like in etcd (enabled by default).
 
 Persistence:
 
@@ -103,7 +110,6 @@ function parse()
             options[arg.substr(2)] = process.argv[++i];
         }
     }
-    options['stale_read'] = options['stale_read'] === '1' || options['stale_read'] === 'yes' || options['stale_read'] === 'true';
     if (options['persist_filter'])
     {
         options['persist_filter'] = require(options['persist_filter'])(options);

antietcd.js

@@ -15,13 +15,18 @@ const ws = require('ws');
 const EtcTree = require('./etctree.js');
 const AntiPersistence = require('./antipersistence.js');
 const AntiCluster = require('./anticluster.js');
-const { runCallbacks, RequestError } = require('./common.js');
+const { runCallbacks, de64, b64, RequestError } = require('./common.js');
+
+const VERSION = '1.1.1';
 
 class AntiEtcd extends EventEmitter
 {
     constructor(cfg)
     {
         super();
+        cfg['merge_watches'] = !('merge_watches' in cfg) || is_true(cfg['merge_watches']);
+        cfg['stale_read'] = !('stale_read' in cfg) || is_true(cfg['stale_read']);
+        cfg['use_base64'] = !('use_base64' in cfg) || is_true(cfg['use_base64']);
         this.clients = {};
         this.client_id = 1;
         this.etctree = new EtcTree(true);
@@ -72,10 +77,10 @@ class AntiEtcd extends EventEmitter
         {
             this.server = http.createServer((req, res) => this._handleRequest(req, res));
         }
-        this.wss = new ws.WebSocketServer({ server: this.server });
+        this.wss = new ws.Server({ server: this.server });
         // eslint-disable-next-line no-unused-vars
         this.wss.on('connection', (conn, req) => this._startWebsocket(conn, null));
-        this.server.listen(this.cfg.port || 2379);
+        this.server.listen(this.cfg.port || 2379, this.cfg.ip || undefined);
     }
 
     async stop()
@@ -118,7 +123,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();
@@ -132,11 +137,11 @@ class AntiEtcd extends EventEmitter
         if (!this.cluster)
         {
             // Run deletion compaction without followers
-            const mod_revision = this.antietcd.etctree.mod_revision;
-            if (mod_revision - this.antietcd.etctree.compact_revision > (this.cfg.compact_revisions||1000)*2)
+            const mod_revision = this.etctree.mod_revision;
+            if (mod_revision - this.etctree.compact_revision > (this.cfg.compact_revisions||1000)*2)
             {
                 const revision = mod_revision - (this.cfg.compact_revisions||1000);
-                this.antietcd.etctree.compact(revision);
+                this.etctree.compact(revision);
             }
         }
     }
@@ -181,13 +186,13 @@ class AntiEtcd extends EventEmitter
                 if (e instanceof RequestError)
                 {
                     code = e.code;
-                    reply = e.message;
+                    reply = e.message+'\n';
                 }
                 else
                 {
                     console.error(e);
                     code = 500;
-                    reply = 'Internal error: '+e.message;
+                    reply = 'Internal error: '+e.message+'\n';
                 }
             }
             try
@@ -329,7 +334,7 @@ class AntiEtcd extends EventEmitter
             if ((e instanceof RequestError) && e.code == 404)
             {
                 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');
+                    '/v3/lease/grant, /v3/lease/revoke, /v3/kv/lease/revoke, /v3/lease/keepalive, /v3/maintenance/status');
             }
             else
             {
@@ -345,7 +350,7 @@ class AntiEtcd extends EventEmitter
         {
             throw new RequestError(502, 'Server is stopping');
         }
-        if (path !== 'dump' && this.cluster)
+        if (this.cluster && path !== 'dump' && path != 'maintenance_status')
         {
             const res = await this.cluster.checkRaftState(
                 path,
@@ -409,9 +414,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(this._encodeWatch(params), (msg) => callback(this._encodeMsg(msg)), stream_id);
         if (!watch.created)
         {
             throw new RequestError(400, 'Requested watch revision is compacted', { compact_revision: watch.compact_revision });
@@ -428,31 +433,71 @@ class AntiEtcd extends EventEmitter
         {
             throw new RequestError(400, 'Watch not found');
         }
-        this.etctree.api_cancel_watch({ watch_id: mapped_id });
+        this.etctree.api_cancel_watch(mapped_id);
         delete this.api_watches[watch_id];
     }
 
     // internal handlers
     async _handle_kv_txn(data)
     {
-        return await this.etctree.api_txn(data);
+        if (this.cfg.use_base64)
+        {
+            for (const item of data.compare||[])
+            {
+                if (item.key != null)
+                    item.key = de64(item.key);
+            }
+            for (const items of [ data.success, data.failure ])
+            {
+                for (const item of items||[])
+                {
+                    const req = item.request_range || item.requestRange ||
+                        item.request_put || item.requestPut ||
+                        item.request_delete_range || item.requestDeleteRange;
+                    if (req.key != null)
+                        req.key = de64(req.key);
+                    if (req.range_end != null)
+                        req.range_end = de64(req.range_end);
+                    if (req.value != null)
+                        req.value = de64(req.value);
+                }
+            }
+        }
+        const result = await this.etctree.api_txn(data);
+        if (this.cfg.use_base64)
+        {
+            for (const item of result.responses||[])
+            {
+                if (item.response_range)
+                {
+                    for (const kv of item.response_range.kvs)
+                    {
+                        if (kv.key != null)
+                            kv.key = b64(kv.key);
+                        if (kv.value != null)
+                            kv.value = b64(kv.value);
+                    }
+                }
+            }
+        }
+        return result;
     }
 
     async _handle_kv_range(data)
     {
-        const r = await this.etctree.api_txn({ success: [ { request_range: data } ] });
+        const r = await this._handle_kv_txn({ success: [ { request_range: data } ] });
         return { header: r.header, ...r.responses[0].response_range };
     }
 
     async _handle_kv_put(data)
     {
-        const r = await this.etctree.api_txn({ success: [ { request_put: data } ] });
+        const r = await this._handle_kv_txn({ success: [ { request_put: data } ] });
         return { header: r.header, ...r.responses[0].response_put };
     }
 
     async _handle_kv_deleterange(data)
     {
-        const r = await this.etctree.api_txn({ success: [ { request_delete_range: data } ] });
+        const r = await this._handle_kv_txn({ success: [ { request_delete_range: data } ] });
         return { header: r.header, ...r.responses[0].response_delete_range };
     }
 
@@ -476,6 +521,26 @@ class AntiEtcd extends EventEmitter
         return this.etctree.api_keepalive_lease(data);
     }
 
+    _handle_maintenance_status(/*data*/)
+    {
+        const raft = this.cluster && this.cluster.raft;
+        return {
+            header: {
+                member_id: this.cfg.node_id || undefined,
+                revision: this.etctree.mod_revision,
+                compact_revision: this.etctree.compact_revision || 0,
+                raft_term: raft && raft.term || undefined,
+            },
+            version: 'antietcd '+AntiEtcd.VERSION,
+            cluster: this.cfg.cluster || undefined,
+            leader: raft && raft.leader || undefined,
+            followers: raft && raft.followers || undefined,
+            raftTerm: raft && raft.term || undefined,
+            raftState: raft && raft.state || undefined,
+            dbSize: process.memoryUsage().heapUsed,
+        };
+    }
+
     // eslint-disable-next-line no-unused-vars
     _handle_dump(data)
     {
@@ -494,8 +559,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(this._encodeMsg(msg))));
                 const watch = this.etctree.api_create_watch(
-                    { ...create_request, watch_id: null }, (msg) => socket.send(JSON.stringify(msg))
+                    this._encodeWatch(create_request), client.send_cb, (this.cfg.merge_watches ? 'C'+client_id : null)
                 );
                 if (!watch.created)
                 {
@@ -514,7 +580,7 @@ class AntiEtcd extends EventEmitter
             const mapped_id = client.watches[msg.cancel_request.watch_id];
             if (mapped_id)
             {
-                this.etctree.api_cancel_watch({ watch_id: mapped_id });
+                this.etctree.api_cancel_watch(mapped_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 } }));
             }
@@ -533,6 +599,35 @@ class AntiEtcd extends EventEmitter
         }
     }
 
+    _encodeWatch(create_request)
+    {
+        const req = { ...create_request, watch_id: null };
+        if (this.cfg.use_base64)
+        {
+            if (req.key != null)
+                req.key = de64(req.key);
+            if (req.range_end != null)
+                req.range_end = de64(req.range_end);
+        }
+        return req;
+    }
+
+    _encodeMsg(msg)
+    {
+        if (this.cfg.use_base64 && msg.result && msg.result.events)
+        {
+            return { ...msg, result: { ...msg.result, events: msg.result.events.map(ev => ({
+                ...ev,
+                kv: !ev.kv ? ev.kv : {
+                    ...ev.kv,
+                    key: b64(ev.kv.key),
+                    value: b64(ev.kv.value),
+                },
+            })) } };
+        }
+        return msg;
+    }
+
     _unsubscribeClient(client_id)
     {
         if (!this.clients[client_id])
@@ -542,11 +637,18 @@ class AntiEtcd extends EventEmitter
         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 });
+            this.etctree.api_cancel_watch(mapped_id);
         }
     }
 }
 
+function is_true(s)
+{
+    return s === true || s === 1 || s === '1' || s === 'yes' || s === 'true' || s === 'on';
+}
+
 AntiEtcd.RequestError = RequestError;
 
+AntiEtcd.VERSION = VERSION;
+
 module.exports = AntiEtcd;

antipersistence.js

@@ -8,7 +8,7 @@ const zlib = require('zlib');
 
 const stableStringify = require('./stable-stringify.js');
 const EtcTree = require('./etctree.js');
-const { de64, runCallbacks } = require('./common.js');
+const { runCallbacks } = require('./common.js');
 
 class AntiPersistence
 {
@@ -60,20 +60,18 @@ class AntiPersistence
                 if (ev.lease)
                 {
                     // Values with lease are never persisted
-                    const key = de64(ev.key);
-                    if (this.prev_value[key] !== undefined)
+                    if (this.prev_value[ev.key] !== undefined)
                     {
-                        delete this.prev_value[key];
+                        delete this.prev_value[ev.key];
                         changed = true;
                     }
                 }
                 else
                 {
-                    const key = de64(ev.key);
-                    const filtered = this.cfg.persist_filter(key, ev.value == null ? undefined : de64(ev.value));
-                    if (!EtcTree.eq(filtered, this.prev_value[key]))
+                    const filtered = this.cfg.persist_filter(ev.key, ev.value == null ? undefined : ev.value);
+                    if (!EtcTree.eq(filtered, this.prev_value[ev.key]))
                     {
-                        this.prev_value[key] = filtered;
+                        this.prev_value[ev.key] = filtered;
                         changed = true;
                     }
                 }
@@ -116,7 +114,7 @@ class AntiPersistence
         this.wait_persist = [];
         try
         {
-            let dump = this.antietcd.etctree.dump(true);
+            let dump = this.antietcd.etctree.dump(true, this.cfg.persist_filter);
             dump['term'] = this.antietcd.stored_term;
             dump = stableStringify(dump);
             dump = await new Promise((ok, no) => zlib.gzip(dump, (err, res) => err ? no(err) : ok(res)));

common.js

@@ -1,10 +1,11 @@
 // (c) Vitaliy Filippov, 2024
 // License: Mozilla Public License 2.0 or Vitastor Network Public License 1.1
 
-class RequestError
+class RequestError extends Error
 {
     constructor(code, text, details)
     {
+        super();
         this.code = code;
         this.message = text;
         this.details = details;
@@ -18,6 +19,13 @@ function de64(k)
     return Buffer.from(k, 'base64').toString();
 }
 
+function b64(k)
+{
+    if (k == null) // null or undefined
+        return k;
+    return Buffer.from(k).toString('base64');
+}
+
 function runCallbacks(obj, key, new_value)
 {
     const cbs = obj[key];
@@ -34,5 +42,6 @@ function runCallbacks(obj, key, new_value)
 module.exports = {
     RequestError,
     de64,
+    b64,
     runCallbacks,
 };

etctree.js

@@ -19,7 +19,7 @@ const { RequestError } = require('./common.js');
 
 class EtcTree
 {
-    constructor(use_base64)
+    constructor()
     {
         this.state = {};
         this.leases = {};
@@ -27,7 +27,6 @@ class EtcTree
         this.watcher_id = 0;
         this.mod_revision = 0;
         this.compact_revision = 0;
-        this.use_base64 = use_base64;
         this.replicate = null;
         this.paused = false;
         this.active_immediate = [];
@@ -51,23 +50,9 @@ class EtcTree
         this.replicate = replicate;
     }
 
-    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(this.de64(chk.key));
+        const parts = this._key_parts(chk.key);
         const { cur } = this._get_subtree(parts, false, false);
         let check_value, ref_value;
         if (chk.target === 'MOD')
@@ -118,9 +103,9 @@ class EtcTree
 
     _get_range(req)
     {
-        const key = this.de64(req.key);
-        const end = this.de64(req.range_end);
-        if (end != null && (key[key.length-1] != '/' || end[end.length-1] != '0' ||
+        const key = req.key;
+        const end = req.range_end;
+        if (end != null && (key !== '' && end !== '') && (key[key.length-1] != '/' || end[end.length-1] != '0' ||
             end.substr(0, end.length-1) !== key.substr(0, key.length-1)))
         {
             throw new RequestError(501, 'Non-directory range queries are unsupported');
@@ -168,7 +153,7 @@ class EtcTree
     dump(persistent_only, value_filter)
     {
         const snapshot = {
-            state: this._copy_tree(this.state, persistent_only, value_filter) || {},
+            state: this._copy_tree(this.state, null, persistent_only, value_filter) || {},
             mod_revision: this.mod_revision,
             compact_revision: this.compact_revision,
         };
@@ -184,13 +169,13 @@ class EtcTree
         return snapshot;
     }
 
-    _copy_tree(cur, no_lease, value_filter)
+    _copy_tree(cur, key, no_lease, value_filter)
     {
         let nonempty = cur.value != null && (!no_lease || !cur.lease);
         let filtered;
         if (nonempty && value_filter)
         {
-            filtered = value_filter(cur.value);
+            filtered = value_filter(key === null ? '' : key, cur.value);
             nonempty = nonempty && filtered != null;
         }
         const copy = (nonempty ? { ...cur } : {});
@@ -204,7 +189,7 @@ class EtcTree
         let has_children = false;
         for (const k in cur.children)
         {
-            const child = this._copy_tree(cur.children[k], no_lease, value_filter);
+            const child = this._copy_tree(cur.children[k], key === null ? k : key+'/'+k, no_lease, value_filter);
             if (child)
             {
                 copy.children[k] = child;
@@ -446,6 +431,11 @@ class EtcTree
         {
             this._delete_range({ key }, next_revision, notifications);
         }
+        if (this.leases[id].timer_id)
+        {
+            clearTimeout(this.leases[id].timer_id);
+            this.leases[id].timer_id = null;
+        }
         delete this.leases[id];
     }
 
@@ -539,13 +529,13 @@ 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)
         {
             // Deletions up to this.compact_revision are forgotten
-            return { compact_revision: this.compact_revision };
+            return { canceled: true, cancel_reason: 'Revisions up to '+this.compact_revision+' are compacted', compact_revision: this.compact_revision };
         }
         let watch_id = req.watch_id;
         if (watch_id instanceof Object)
@@ -561,6 +551,7 @@ class EtcTree
             this.watchers[watch_id] = {
                 paths: [],
                 send,
+                stream_id,
             };
         }
         this.watchers[watch_id].paths.push(parts);
@@ -575,7 +566,7 @@ class EtcTree
             cur.key_watchers = cur.key_watchers || [];
             cur.key_watchers.push(watch_id);
         }
-        if (req.start_revision && req.start_revision < this.mod_revision)
+        if (req.start_revision && req.start_revision <= this.mod_revision)
         {
             // Send initial changes
             const imm = setImmediate(() =>
@@ -588,7 +579,7 @@ class EtcTree
             });
             this.active_immediate.push(imm);
         }
-        return { watch_id, created: true };
+        return { header: { revision: this.mod_revision }, watch_id, created: true };
     }
 
     _get_modified(events, cur, prefix, min_rev)
@@ -597,9 +588,9 @@ class EtcTree
         {
             const ev = {
                 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),
+                kv: cur.value == null ? { key: (prefix === null ? '' : prefix) } : {
+                    key: prefix,
+                    value: cur.value,
                     mod_revision: cur.mod_revision,
                 },
             };
@@ -662,15 +653,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) } });
         }
     }
 
@@ -727,9 +718,9 @@ class EtcTree
     _put(request_put, cur_revision, notifications)
     {
         // FIXME: prev_kv, ignore_value(?), ignore_lease(?)
-        const parts = this._key_parts(this.de64(request_put.key));
+        const parts = this._key_parts(request_put.key);
         const key = parts.join('/');
-        const value = this.de64(request_put.value);
+        const value = request_put.value;
         const { cur, watchers } = this._get_subtree(parts, true, true);
         if (cur.key_watchers)
         {
@@ -762,7 +753,7 @@ class EtcTree
                 cur.create_revision = cur_revision;
             }
             cur.value = value;
-            const notify = { watchers, key: this.b64(key), value: this.b64(value), mod_revision: cur.mod_revision };
+            const notify = { watchers, key, value, mod_revision: cur.mod_revision };
             if (cur.lease)
             {
                 notify.lease = cur.lease;
@@ -793,10 +784,10 @@ class EtcTree
         }
         if (cur.value != null)
         {
-            const item = { key: this.b64(prefix === null ? '' : prefix) };
+            const item = { key: (prefix === null ? '' : prefix) };
             if (!req.keys_only)
             {
-                item.value = this.b64(cur.value);
+                item.value = cur.value;
                 item.mod_revision = cur.mod_revision;
                 //item.create_revision = cur.create_revision;
                 //item.version = cur.version;
@@ -833,7 +824,7 @@ class EtcTree
             this.mod_revision = cur_revision;
             notifications.push({
                 watchers: cur.key_watchers ? [ ...watchers, ...cur.key_watchers ] : watchers,
-                key: this.b64(prefix === null ? '' : prefix),
+                key: (prefix === null ? '' : prefix),
                 mod_revision: cur_revision,
             });
         }

etctree.spec.js

@@ -64,7 +64,7 @@ tests['read/write'] = async () =>
     );
     expect(
         t.dump(false),
-        {"state":{"children":{"":{"children":{"vitastor":{"children":{"config":{"children":{"global":{"version":2,"mod_revision":2,"create_revision":1,"value":{"hello":"world2"}}}}}}}}}},"mod_revision":2,"leases":{}}
+        {"state":{"children":{"":{"children":{"vitastor":{"children":{"config":{"children":{"global":{"version":2,"mod_revision":2,"create_revision":1,"value":{"hello":"world2"}}}}}}}}}},"mod_revision":2,"compact_revision":0,"leases":{}}
     );
     t.destroy();
 };
@@ -80,7 +80,7 @@ tests['watch'] = async () =>
     );
     expect(
         t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send),
-        { watch_id: 1, created: true }
+        { header: { revision: 1 }, watch_id: 1, created: true }
     );
     expect(sent, []);
     expect(
@@ -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();
@@ -100,18 +132,18 @@ tests['lease'] = async () =>
     expect(leaseID != null, true);
     expect(
         await t.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.4' } } } ] }),
-        { header: { revision: 1 }, succeeded: true, responses: [ { response_put: {} } ] }
+        { header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} } ] }
     );
     expect(
         t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send),
-        { watch_id: 1, created: true }
+        { header: { revision: 2 }, watch_id: 1, created: true }
     );
     expect(sent, []);
     const dump = t.dump(false);
     const expires = dump.leases[leaseID].expires;
-    expect(dump, {"state":{"children":{"":{"children":{"vitastor":{"children":{"osd":{"children":{"state":{"children":{"1":{"lease":leaseID,"version":1,"mod_revision":1,"create_revision":1,"value":{"ip":"1.2.3.4"}}}}}}}}}}}},"mod_revision":1,"leases":{[leaseID]:{"ttl":0.5,"expires":expires}}});
+    expect(dump, {"state":{"children":{"":{"children":{"vitastor":{"children":{"osd":{"children":{"state":{"children":{"1":{"lease":leaseID,"version":1,"mod_revision":2,"create_revision":2,"value":{"ip":"1.2.3.4"}}}}}}}}}}}},"mod_revision":2,"compact_revision":0,"leases":{[leaseID]:{"ttl":0.5,"expires":expires}}});
     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 } } ] } } ]);
+    expect(sent, [ { result: { header: { revision: 3 }, events: [ { type: 'DELETE', kv: { key: '/vitastor/osd/state/1', mod_revision: 3 } } ] } } ]);
     t.pause_leases();
     t.load(dump);
     expect(t.dump(false), dump);
@@ -131,7 +163,7 @@ tests['update'] = async () =>
     expect(leaseID != null, true);
     expect(
         await t1.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.4' } } } ] }),
-        { header: { revision: 1 }, succeeded: true, responses: [ { response_put: {} } ] }
+        { header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} } ] }
     );
     expect(
         await t2.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', value: { ip: '1.2.3.6' } } } ] }),
@@ -139,19 +171,38 @@ tests['update'] = async () =>
     );
     expect(
         await t1.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.5' } } } ] }),
-        { header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} } ] }
+        { header: { revision: 3 }, succeeded: true, responses: [ { response_put: {} } ] }
     );
     let dump2 = t2.dump();
     t2.load(t1.dump(), true);
     t1.load(dump2, true);
     let dump = t2.dump(false);
     let expires = dump.leases[leaseID].expires;
-    expect(dump, {"state":{"children":{"":{"children":{"vitastor":{"children":{"osd":{"children":{"state":{"children":{"1":{"lease":leaseID,"version":2,"mod_revision":2,"create_revision":1,"value":{"ip":"1.2.3.5"}}}}}}}}}}}},"mod_revision":2,"leases":{[leaseID]:{"ttl":0.5,"expires":expires}}});
-    expect(t1.dump(false), {"state":{"children":{"":{"children":{"vitastor":{"children":{"osd":{"children":{"state":{"children":{"1":{"lease":leaseID,"version":2,"mod_revision":2,"create_revision":1,"value":{"ip":"1.2.3.5"}}}}}}}}}}}},"mod_revision":2,"leases":{[leaseID]:{"ttl":0.5,"expires":expires}}});
+    expect(dump, {"state":{"children":{"":{"children":{"vitastor":{"children":{"osd":{"children":{"state":{"children":{"1":{"lease":leaseID,"version":2,"mod_revision":3,"create_revision":2,"value":{"ip":"1.2.3.5"}}}}}}}}}}}},"mod_revision":3,"compact_revision":0,"leases":{[leaseID]:{"ttl":0.5,"expires":expires}}});
+    expect(t1.dump(false), {"state":{"children":{"":{"children":{"vitastor":{"children":{"osd":{"children":{"state":{"children":{"1":{"lease":leaseID,"version":2,"mod_revision":3,"create_revision":2,"value":{"ip":"1.2.3.5"}}}}}}}}}}}},"mod_revision":3,"compact_revision":0,"leases":{[leaseID]:{"ttl":0.5,"expires":expires}}});
     t1.destroy();
     t2.destroy();
 };
 
+tests['dump filter'] = async () =>
+{
+    const t1 = new EtcTree();
+    const leaseID = (await t1.api_grant_lease({ TTL: 0.5 })).ID;
+    expect(leaseID != null, true);
+    expect(
+        await t1.api_txn({ success: [
+            { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.6' } } },
+            { request_put: { key: '/vitastor/osd/stats/1', value: { time: 123, read: [ 10, 1000, 10459 ] } } },
+        ] }),
+        { header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} }, { response_put: {} } ] }
+    );
+    expect(
+        t1.dump(true, (key, value) => (key.substr(0, '/vitastor/osd/stats'.length) === '/vitastor/osd/stats' ? { time: value.time } : value)),
+        {"state":{"children":{"":{"children":{"vitastor":{"children":{"osd":{"children":{"stats":{"children":{"1":{"version":1,"mod_revision":2,"create_revision":2,"value":{"time":123}}}}}}}}}}}},"mod_revision":2,"compact_revision":0}
+    );
+    t1.destroy();
+};
+
 tests['replicate watcher'] = async () =>
 {
     const t = new EtcTree();

package.json

@@ -1,6 +1,6 @@
 {
   "name": "antietcd",
-  "version": "1.0.0",
+  "version": "1.1.1",
   "description": "Simplistic etcd replacement based on TinyRaft",
   "main": "antietcd.js",
   "scripts": {

vitastor_persist_filter.js

@@ -24,7 +24,7 @@ function vitastor_persist_filter(cfg)
                 catch (e)
                 {
                     console.error('invalid JSON in '+key+' = '+value+': '+e);
-                    value = {};
+                    value = '{}';
                 }
             }
             else