Support Cloudflare workers

[shiyuhang0]

Recently the Postgres community supported Cloudflare worker with its Socket API. Here are some references:

• node-postgres: https://github.com/brianc/node-postgres/commit/07553428e9c0eacf761a5d4541a3300ff7859578
• Postgres.js: https://github.com/porsager/postgres/commit/b88e261b7625f4659cae17197b395ad6933732f1

One of the benefits is to connect to the Cloudflare Worker via the TCP connection in MySQL ecosystem. And I think it may not be limited to Cloudflare Works, it may also work in other edge runtimes in the future, e.g. vercel edge function.

Thus, I want to discuss if it is possible to support Cloudflare Worker in mysql2. Thanks!

[sidorares]

should be possible, I expect implementation to be close to node-postgres

Getting tls to work will be a bit trickier, I believe postgres optionally connects using tls from the very beginning so the api is "getStream / getSecureStream". Mysql on the other hand always start a connection without tls and then upgrades same tcp stream to use tls

You can pass any duplex stream via stream. If you want to try to create POC I suggest trying to wrap cloudflare socket and pass it as a stream

[shiyuhang0]

I am new to javascript, but i will have a try

[shiyuhang0]

This commit https://github.com/brianc/node-postgres/commit/5532ca51db96f3370faf66d9e13f0ba226844f62#diff-fcad7be0755aac2e767a566a1e589de80634c52168f34bc555c6881d41244f24 replace node's crypto API to WebCrypto APIs because Cloudflare works does not have crypto AP. This makes the corresponding function an async function.

But in node-mysql2, it causes a large number of functions to be async. Maybe it is not a good choice in node-mysql2. Does anyone have some suggestions?

[benycodes]

Cloudflare Socket wrapper:

import { EventEmitter } from 'events';
/**
 * Wrapper around the Cloudflare built-in socket that can be used by the `Connection`.
 */
export class CloudflareSocket extends EventEmitter {
    constructor(ssl) {
        super();
        this.ssl = ssl;
        this.writable = false;
        this.destroyed = false;
        this._upgrading = false;
        this._upgraded = false;
        this._cfSocket = null;
        this._cfWriter = null;
        this._cfReader = null;
    }
    setNoDelay() {
        return this;
    }
    setKeepAlive() {
        return this;
    }
    ref() {
        return this;
    }
    unref() {
        return this;
    }
    async connect(port, host, connectListener) {
        try {
            log('connecting');
            if (connectListener)
                this.once('connect', connectListener);
            const options = this.ssl ? { secureTransport: 'starttls' } : {};
            const { connect } = await import('cloudflare:sockets');
            this._cfSocket = connect(`${host}:${port}`, options);
            this._cfWriter = this._cfSocket.writable.getWriter();
            this._addClosedHandler();
            this._cfReader = this._cfSocket.readable.getReader();
            if (this.ssl) {
                this._listenOnce().catch((e) => this.emit('error', e));
            }
            else {
                this._listen().catch((e) => this.emit('error', e));
            }
            await this._cfWriter.ready;
            log('socket ready');
            this.writable = true;
            this.emit('connect');
            return this;
        }
        catch (e) {
            this.emit('error', e);
        }
    }
    async _listen() {
        while (true) {
            log('awaiting receive from CF socket');
            const { done, value } = await this._cfReader.read();
            log('CF socket received:', done, value);
            if (done) {
                log('done');
                break;
            }
            this.emit('data', Buffer.from(value));
        }
    }
    async _listenOnce() {
        log('awaiting first receive from CF socket');
        const { done, value } = await this._cfReader.read();
        log('First CF socket received:', done, value);
        this.emit('data', Buffer.from(value));
    }
    write(data, encoding = 'utf8', callback = () => { }) {
        if (data.length === 0)
            return callback();
        if (typeof data === 'string')
            data = Buffer.from(data, encoding);
        log('sending data direct:', data);
        this._cfWriter.write(data).then(() => {
            log('data sent');
            callback();
        }, (err) => {
            log('send error', err);
            callback(err);
        });
        return true;
    }
    end(data = Buffer.alloc(0), encoding = 'utf8', callback = () => { }) {
        log('ending CF socket');
        this.write(data, encoding, (err) => {
            this._cfSocket.close();
            if (callback)
                callback(err);
        });
        return this;
    }
    destroy(reason) {
        log('destroying CF socket', reason);
        this.destroyed = true;
        return this.end();
    }
    startTls(options) {
        if (this._upgraded) {
            // Don't try to upgrade again.
            this.emit('error', 'Cannot call `startTls()` more than once on a socket');
            return;
        }
        this._cfWriter.releaseLock();
        this._cfReader.releaseLock();
        this._upgrading = true;
        this._cfSocket = this._cfSocket.startTls(options);
        this._cfWriter = this._cfSocket.writable.getWriter();
        this._cfReader = this._cfSocket.readable.getReader();
        this._addClosedHandler();
        this._listen().catch((e) => this.emit('error', e));
    }
    _addClosedHandler() {
        this._cfSocket.closed.then(() => {
            if (!this._upgrading) {
                log('CF socket closed');
                this._cfSocket = null;
                this.emit('close');
            }
            else {
                this._upgrading = false;
                this._upgraded = true;
            }
        }).catch((e) => this.emit('error', e));
    }
}
const debug = false;
function dump(data) {
    if (data instanceof Uint8Array || data instanceof ArrayBuffer) {
        const hex = Buffer.from(data).toString('hex');
        const str = new TextDecoder().decode(data);
        return `\n>>> STR: "${str.replace(/\n/g, '\\n')}"\n>>> HEX: ${hex}\n`;
    }
    else {
        return data;
    }
}
function log(...args) {
    debug && console.log(...args.map(dump));
}

[sidorares]

@benycodes do you want to publish this as a standalone package? And we could mention example in mysql2 documentation

[benycodes]

@benycodes do you want to publish this as a standalone package? And we could mention example in mysql2 documentation

I've been working on it more, but seems like there is more than just the socket. The code used in the password hashing doesn't work on Cloudflare Workers either. It needs to be done like this import { createHash } from 'node:crypto'; instead of const crypto = require('crypto');

Will publish when I have something working.

[shiyuhang0]

Why not use https://www.npmjs.com/package/pg-cloudflare directly?

[shiyuhang0]

@benycodes do you want to publish this as a standalone package? And we could mention example in mysql2 documentation

I've been working on it more, but seems like there is more than just the socket. The code used in the password hashing doesn't work on Cloudflare Workers either. It needs to be done like this import { createHash } from 'node:crypto'; instead of const crypto = require('crypto');

Will publish when I have something working.

Nice catch! Just found Cloudflare supports crypto last month. I used WebCrypto API before which increased a lot of work.

[shiyuhang0]

@benycodes do you want to publish this as a standalone package? And we could mention example in mysql2 documentation

I've been working on it more, but seems like there is more than just the socket. The code used in the password hashing doesn't work on Cloudflare Workers either. It needs to be done like this import { createHash } from 'node:crypto'; instead of const crypto = require('crypto');

Will publish when I have something working.

Create a pr https://github.com/shiyuhang0/node-mysql2/pull/1 but stills have the following two problems

replacing crypto

1. using import { createHash } from 'node:crypto' in workers needs compatibility_flags = [ "nodejs_compat" ] flag. This flag is conflicted with node_compat = true flag. I found some compatibility issues occur without node_compat = true flag. I think it's hard to overcome because some of them are mysql2 dependencies. Are you also facing the same problem?

2. using WebCrypto API makes a large number of functions to be async. fail to make it working now :(

Code generation from strings disallowed for this context

connect to a user without a password to avoid calling crypto. But the workers return Code generation from strings disallowed for this context. The stack points to https://github.com/sidorares/node-mysql2/blob/1aec4fdb856204f8e4f277eb3fc4526f29c5af90/promise.js#L112

The above two problems blocks me now.

[Mini256]

Because of security restrictions,text_parser.js will throw EvalError in the CloudFlare worker environment:

EvalError: Code generation from strings disallowed for this context
    at Function (<anonymous>)
    at line.toFunction (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:25455:25)
    at compile (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:25680:23)
    at Object.getParser (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:25492:16)
    at getTextParser (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:25683:26)
    at Query.readField (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:25882:34)
    at Query.execute (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:19727:26)
    at Connection2.handlePacket (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:30310:39)
    at PacketParser.onPacket (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:29997:16)
    at PacketParser.executeStart (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:17284:20) {
  fatal: true
}

[shiyuhang0]

Because of security restrictions,text_parser.js will throw EvalError in the CloudFlare worker environment:

EvalError: Code generation from strings disallowed for this context
    at Function (<anonymous>)
    at line.toFunction (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:25455:25)
    at compile (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:25680:23)
    at Object.getParser (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:25492:16)
    at getTextParser (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:25683:26)
    at Query.readField (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:25882:34)
    at Query.execute (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:19727:26)
    at Connection2.handlePacket (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:30310:39)
    at PacketParser.onPacket (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:29997:16)
    at PacketParser.executeStart (file:///Users/liangzhiyuan/playground/mysql2-cloudflare-test/.wrangler/tmp/dev-WCcN3p/index.js:17284:20) {
  fatal: true
}

I don't know if it is possible to refactor this part without generate-function. @sidorares what's your opinion?

[sidorares]

I have plans to add "non-eval" parser, initially due to perf reasons ( when number of fields is too big it might be better to use for loop + field type check ). If we add an option like maxNumFieldsGenerateFunction that would make it possible to force 'interpreting' parser by setting it to 0

[Mini256]

@sidorares I copied the interpreting parser in PR #2099 and tested it locally, which seems to work properly.

But I'm not sure yet how much performance difference the static parser has compared to the existed parser.

export default {
    async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
        try {
          const tidbConn = createConnection({
                  host: '127.0.0.1',
                  port: 4000,
                  user: 'root',
                  password: '',
                  database: 'test',
                  debug: true,
                  useStaticParser: true,
          });
          // @ts-ignore
          const [rows] = await query(tidbConn, 'SELECT 1 AS field;');
          tidbConn.end();
          return new Response(JSON.stringify(rows), { status: 200 });
        } catch (e: any) {
          console.error(e);
          return new Response('Error: ' + e.message, { status: 500 });
        }
    },
};

[sidorares]

But I'm not sure yet how much performance difference the static parser has compared to the existed parser.

depends on the load, "JIT" parser works best on a large result set ( hundreds to a millions of rows ) and small to a medium number of fields ( 5 -50 ), can't remember exactly but I believe the difference in double digits of percents