anticli.js

@@ -83,7 +83,6 @@ class AntiEtcdCli
         {
             await this.del(cmd.slice(1));
         }
-        process.exit(0);
     }
 
     async get(keys)
@@ -94,7 +93,7 @@ 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);
-        for (const r of res.responses||[])
+        for (const r of (res||{}).responses||[])
         {
             if (r.response_range)
             {
@@ -120,7 +119,7 @@ class AntiEtcdCli
             value = await fsp.readFile(0, { encoding: 'utf-8' });
         }
         const res = await this.request('/v3/kv/put', { key: b64(key), value: b64(value) });
-        if (res.header)
+        if (res && res.header)
         {
             console.log('OK');
         }
@@ -134,7 +133,7 @@ class AntiEtcdCli
         }
         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||[])
+        for (const r of (res||{}).responses||[])
         {
             if (r.response_delete_range)
             {
@@ -148,33 +147,16 @@ class AntiEtcdCli
         for (const url of this.options.endpoints)
         {
             const cur_url = url.replace(/\/+$/, '')+path;
-            const res = await POST(cur_url, body, this.options.timeout||1000);
-            let ok = false;
-            if (res.json)
+            try
             {
-                if (res.json.error)
+                return (await POST(cur_url, body, this.options.timeout||1000)).json;
+            }
+            catch (e)
             {
-                    console.error(cur_url+': '+res.json.error);
-                    process.exit(1);
-                }
-                return res.json;
-            }
-            if (res.body)
-            {
-                console.error(cur_url+': '+res.body);
-            }
-            if (res.error)
-            {
-                console.error(cur_url+': '+res.error);
-                if (!res.response || !res.response.statusCode)
-                {
-                    // This URL is unavailable
-                    continue;
+                console.error(cur_url+': '+e.message);
             }
         }
-            break;
-        }
-        process.exit(1);
+        return null;
     }
 }
 
@@ -206,9 +188,9 @@ function POST(url, body, timeout)
             res.on('data', chunk => { res_body += chunk; });
             res.on('end', () =>
             {
-                if (res.statusCode != 200 || !/application\/json/i.exec(res.headers['content-type']))
+                if (res.statusCode != 200)
                 {
-                    ok({ response: res, body: res_body, code: res.statusCode });
+                    ok({ error: res_body, code: res.statusCode });
                     return;
                 }
                 try
@@ -218,7 +200,7 @@ function POST(url, body, timeout)
                 }
                 catch (e)
                 {
-                    ok({ response: res, error: e, body: res_body });
+                    ok({ error: e, response: res, body: res_body });
                 }
             });
         });

anticluster.js

@@ -59,7 +59,6 @@ class AntiCluster
                     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 } }));
-                    this.raft.start();
                 }
             });
         }

antietcd.js

@@ -86,29 +86,13 @@ class AntiEtcd
 
     async persistAndReplicate(msg)
     {
-        let res = [];
-        if (this.cluster)
-        {
-            // We have to guarantee that replication is processed sequentially
-            // So we have to send messages without first awaiting for anything!
-            res.push(this.cluster.replicateChange(msg));
-        }
         if (this.persistence)
         {
-            res.push(this.persistence.persistChange(msg));
+            await this.persistence.persistChange(msg);
         }
-        if (res.length)
+        if (this.cluster)
         {
-            res = await Promise.allSettled(res);
-            const errors = res.filter(r => r.status == 'rejected');
-            if (errors.length)
-            {
-                for (const e of errors)
-                {
-                    console.error(e.reason);
-                }
-                throw errors[0].reason;
-            }
+            await this.cluster.replicateChange(msg);
         }
     }
 
@@ -122,7 +106,6 @@ class AntiEtcd
             data = Buffer.concat(data);
             let body = '';
             let code = 200;
-            let ctype = 'text/plain; charset=utf-8';
             let reply;
             try
             {
@@ -145,7 +128,6 @@ class AntiEtcd
                 }
                 reply = await this.runHandler(req, data, res);
                 reply = JSON.stringify(reply);
-                ctype = 'application/json';
             }
             catch (e)
             {
@@ -171,8 +153,8 @@ class AntiEtcd
                     '\n  '+reply.replace(/\n/g, '\\n')
                 );
                 reply = Buffer.from(reply);
-                res.writeHead(code, {
-                    'Content-Type': ctype,
+                res.writeHead(200, {
+                    'Content-Type': 'application/json',
                     'Content-Length': reply.length,
                 });
                 res.write(reply);

antipersistence.js

@@ -2,7 +2,6 @@ const fs = require('fs');
 const fsp = require('fs').promises;
 const zlib = require('zlib');
 
-const stableStringify = require('./stable-stringify.js');
 const EtcTree = require('./etctree.js');
 const { de64, runCallbacks } = require('./common.js');
 
@@ -53,10 +52,10 @@ class AntiPersistence
             let changed = false;
             for (const ev of msg.events)
             {
-                if (ev.lease)
+                if (ev.kv.lease)
                 {
                     // Values with lease are never persisted
-                    const key = de64(ev.key);
+                    const key = de64(ev.kv.key);
                     if (this.prev_value[key] !== undefined)
                     {
                         delete this.prev_value[key];
@@ -65,14 +64,15 @@ class AntiPersistence
                 }
                 else
                 {
-                    const key = de64(ev.key);
-                    const filtered = this.cfg.persist_filter(key, ev.value == null ? undefined : de64(ev.value));
+                    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)
             {
@@ -114,7 +114,7 @@ class AntiPersistence
         {
             let dump = this.antietcd.etctree.dump(true);
             dump['term'] = this.antietcd.stored_term;
-            dump = stableStringify(dump);
+            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);
@@ -124,7 +124,7 @@ class AntiPersistence
         }
         catch (e)
         {
-            console.error('Error persisting data to disk: '+e);
+            console.error(e);
             process.exit(1);
         }
         runCallbacks(this, 'wait_persist', null);

etctree.js

@@ -232,7 +232,7 @@ class EtcTree
                 if (!snapshot.leases[id])
                 {
                     // Revoke without replicating and notifying
-                    this._sync_revoke_lease(id, notifications, this.mod_revision);
+                    this._sync_revoke_lease(id, notifications);
                 }
             }
         }
@@ -269,10 +269,10 @@ class EtcTree
             {
                 cur_old.value = cur_new.value;
                 const key_watchers = (cur_old.key_watchers ? [ ...watchers, ...(cur_old.key_watchers||[]) ] : watchers);
-                const notify = { watchers: key_watchers, key, value: cur_new.value, mod_revision: cur_new.mod_revision };
-                if (cur_new.lease)
+                const notify = { watchers: key_watchers, key, value: cur_old.value, mod_revision: cur_old.mod_revision };
+                if (cur_old.lease)
                 {
-                    notify.lease = cur_new.lease;
+                    notify.lease = cur_old.lease;
                 }
                 notifications.push(notify);
             }
@@ -366,7 +366,6 @@ class EtcTree
         }
         const expires = Date.now() + req.TTL*1000;
         this.leases[id] = { ttl: req.TTL, expires, timer_id: null, keys: {} };
-        this.mod_revision++;
         this._set_expire(id);
         if (this.replicate)
         {
@@ -390,7 +389,6 @@ class EtcTree
         }
         const ttl = this.leases[id].ttl;
         lease.expires = Date.now() + ttl*1000;
-        this.mod_revision++;
         this._set_expire(id);
         if (this.replicate)
         {
@@ -426,12 +424,13 @@ class EtcTree
         return true;
     }
 
-    _sync_revoke_lease(id, notifications, next_revision)
+    _sync_revoke_lease(id, notifications)
     {
         if (!this.leases[id])
         {
             throw new Error('unknown lease');
         }
+        const next_revision = this.mod_revision + 1;
         for (const key in this.leases[id].keys)
         {
             this._delete_range({ key }, next_revision, notifications);
@@ -448,8 +447,7 @@ class EtcTree
                 return null;
             throw new Error('unknown lease');
         }
-        this.mod_revision++;
-        this._sync_revoke_lease(req.ID, notifications, this.mod_revision);
+        this._sync_revoke_lease(req.ID, notifications);
         if (this.replicate)
         {
             await this.notify_replicator(notifications, [ { id: req.ID } ]);
@@ -472,7 +470,6 @@ class EtcTree
 
     async apply_replication(msg)
     {
-        this.mod_revision = msg.header.revision;
         const notifications = [];
         if ((msg.leases||[]).length)
         {
@@ -484,7 +481,7 @@ class EtcTree
                 }
                 else
                 {
-                    this._sync_revoke_lease(lease.id, notifications, this.mod_revision);
+                    this._sync_revoke_lease(lease.id, notifications);
                 }
             }
         }
@@ -498,7 +495,7 @@ class EtcTree
                 }
                 else
                 {
-                    this._put({ key: ev.key, value: ev.value, lease: ev.lease }, ev.mod_revision, notifications);
+                    this._put({ key: ev.key, value: ev.value }, ev.mod_revision, notifications);
                 }
             }
         }