etcd-bug-64k.js

const http = require('http');
const os = require('os');
const WebSocket = require('ws');

class Test
{
    constructor(etcd_url)
    {
        this.etcd_url = etcd_url;
    }

    async run()
    {
        await new Promise((ok, no) =>
        {
            this.ws = new WebSocket(this.etcd_url+'/watch');
            this.ws.on('open', () =>
            {
                ok(true);
            });
        });
        console.log('>> Websocket opened');
        await new Promise((ok, no) =>
        {
            this.ws.once('message', (msg) =>
            {
                console.log(JSON.parse(msg));
                console.log('>> Watch created');
                ok();
            });
            this.ws.send(JSON.stringify({
                create_request: {
                    key: b64('/bug64k/'),
                    range_end: b64('/bug64k0'),
                    watch_id: 1,
                },
            }));
        });
        console.log('>> Writing key /bug64k/test with 1kb value. We should get it back via the websocket');
        await this.test_wr(1024);
        console.log('>> Writing key /bug64k/test with 128kb value. We should get it back via the websocket');
        await this.test_wr(128*1024);
    }

    test_wr(n)
    {
        return new Promise((ok, no) =>
        {
            let fin = 0;
            this.ws.once('message', (msg) =>
            {
                console.log(JSON.parse(msg));
                console.log('>> Got it back, OK');
                fin++;
                if (fin == 2)
                    ok();
            });
            setTimeout(() =>
            {
                console.log('>> Didn\'t get the value back :-(');
                process.exit();
            }, 5000);
            this.etcd_call('/kv/txn', { success: [
                { requestPut: { key: b64('/bug64k/test'), value: b64(Date.now()+' '+Buffer.alloc(n, 'X').toString()) } }
            ] }, 1000, -1).then(() =>
            {
                console.log('>> Key written');
                fin++;
                if (fin == 2)
                    ok();
            }).catch(console.error);
        });
    }

    async etcd_call(path, body, timeout, retries)
    {
        let retry = 0;
        if (retries >= 0 && retries < 1)
        {
            retries = 1;
        }
        while (retries < 0 || retry < retries)
        {
            const res = await POST(this.etcd_url+path, body, timeout);
            if (res.error)
            {
                console.log('etcd returned error: '+res.error);
                break;
            }
            if (res.json)
            {
                if (res.json.error)
                {
                    console.log('etcd returned error: '+res.json.error);
                    break;
                }
                return res.json;
            }
            retry++;
        }
        this.die();
    }
}

function POST(url, body, timeout)
{
    return new Promise((ok, no) =>
    {
        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 = 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('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.write(body_text);
        req.end();
    });
}

function b64(str)
{
    return Buffer.from(str).toString('base64');
}

function de64(str)
{
    return Buffer.from(str, 'base64').toString();
}

(new Test(process.argv[2] || 'http://127.0.0.1:2379/v3')).run().catch(console.error);