Implement inbound UDP handler

[ThomasRubini]

Summary

This PR adds support for workerd to handle inbound UDP "connections" and deliver them to a connect() handler.

It does 2 things:

• adds the protocol field to the Socket object used in connect() handlers, following [wrangler] add connect wrangler config section workers-sdk#14995
• adds a UDP listener to trigger that connect() handler for UDP connections, grouping UDP packets by the same source IP+port under the same connection, with a timeout.

To prevent one "UDP connection"'s slow reader from blocking others, we always read all UDP packets from the kernel, and store them in our own per-connection buffers (in Flow), dropping new packets for a connection when its buffer is full.

The UDP connect() handler has been implemented as a custom event because connect() on WorkerInterface required a kj::AsyncIoStreamwhich is byte-oriented rather than message-oriented like UDP requires

UDP sockets will read and write custom Datagram objects instead of Uint8Array, to avoid user confusion if they try to use them with code that expect stream-based sockets.

Example usage

export default {
  async connect(socket): Promise<void> {
		console.log("New connection");
		const reader = socket.readable.getReader();
		const writer = socket.writable.getWriter();

		const { remoteAddress } = await socket.opened;
		const greeting = new TextEncoder().encode(
			`hello from connect(), protocol: ${socket.protocol}, remoteAddress: ${remoteAddress ?? "unknown"}\n`,
		);
		await writer.write(socket.protocol === "udp" ? new Datagram(greeting) : greeting);

		const decoder = new TextDecoder('utf-8');
		try {
			for (;;) {
				const { value, done } = await reader.read();
				if (done) {
					break;
				}
				console.log("Received datagram: " +  decoder.decode(value.data));
				await writer.write(value);
			}
		} finally {
			await writer.close();
		}
	},
} satisfies ExportedHandler<Env>;

testable by adding this bit in a workerd config:

sockets = [
    ( name = "http", address = "*:8787", http = (), service = "main" ),
    ( name = "tcp", address = "*:5432", tcp = (), service = "main" ),
    ( name = "udp", address = "*:5599", udp = (idleTimeoutMs = 30000, maxPendingBytes = 262144), service = "main" ),
  ]

[github-actions]

All contributors have signed the CLA ✍️ ✅
_{Posted by the CLA Assistant Lite bot.}

[ThomasRubini]

I have read the CLA Document and I hereby sign the CLA

[dom96]

I would ideally like to see this added to the sockets spec before it is implemented: https://sockets-api.proposal.wintertc.org/

It's also worth considering how our design compares to the direct-socket API. Ideally we shouldn't diverge from it unless it's necessary. It would be nice to reuse its UDPMessage + ReadableStream/WritableStream approach for example.

[harrishancock]

I'm still reading, but wanted to leave some initial comments.

[jasnell]

Internal pipeline is green. Merging