Firebase firestore local emulator crash causes functions emulator work queue to fill up and stall indefinitely

### [REQUIRED] Environment info

**firebase-tools:** 11.25.2

**Platform:** OSX

### [REQUIRED] Test case

Crash/stalling behavior appears to happen more often when a number of concurrent requests are made from functions emulator to the firestore emulator.

### [REQUIRED] Steps to reproduce

```
firebase emulators:start
```

### [REQUIRED] Expected behavior

Functions emulator workers should catch emulator-internal errors and trigger userspace functions reliably

### [REQUIRED] Actual behavior

```
[debug] [2023-04-03T19:07:45.418Z] 12:07:45.409 [TriggersWorkerPool-3-2] ERROR com.firebase.server.client.GCFEmulatorPublisher - rtdb failed to publish trigger to functions emulator with exception
com.firebase.server.client.ChannelClosedException$: The underlying socket was closed and released from the connection pool
at com.firebase.server.client.ChannelClosedException$.(SingleHostChannelPool.scala)
at com.firebase.server.client.NettyHttpBesieger.$anonfun$channelCreated$2(NettyHttpBesieger.scala:165)
at com.firebase.server.client.NettyHttpBesieger.$anonfun$channelCreated$2$adapted(NettyHttpBesieger.scala:164)
at com.firebase.server.util.package$NettyFutureAdapter.$anonfun$onComplete$1(package.scala:85)
at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:507)
at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:481)
at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:420)
at io.netty.util.concurrent.DefaultPromise.trySuccess(DefaultPromise.java:104)
at io.netty.channel.DefaultChannelPromise.trySuccess(DefaultChannelPromise.java:82)
at io.netty.channel.AbstractChannel$CloseFuture.setClosed(AbstractChannel.java:1002)
at io.netty.channel.AbstractChannel$AbstractUnsafe.doClose0(AbstractChannel.java:632)
at io.netty.channel.AbstractChannel$AbstractUnsafe.close(AbstractChannel.java:610)
at io.netty.channel.AbstractChannel$AbstractUnsafe.close(AbstractChannel.java:553)
at io.netty.channel.DefaultChannelPipeline$HeadContext.close(DefaultChannelPipeline.java:1244)
at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:618)
at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:602)
at io.netty.channel.ChannelDuplexHandler.close(ChannelDuplexHandler.java:73)
at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:618)
at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:602)
at io.netty.channel.ChannelOutboundHandlerAdapter.close(ChannelOutboundHandlerAdapter.java:71)
at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:618)
at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:602)
at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:459)
at com.firebase.server.client.PipelinedResponseReader.channelRead0(NettyHttpBesieger.scala:292)
at com.firebase.server.client.PipelinedResponseReader.channelRead0(NettyHttpBesieger.scala:209)
at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:102)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:312)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:286)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286)
at com.firebase.server.router.IdleStateDetectionHandler.channelRead(IdleStateDetectionHandler.scala:39)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1302)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:646)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysPlain(NioEventLoop.java:546)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:500)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:460)
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:131)
at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:138)
at java.base/java.lang.Thread.run(Thread.java:1589)
12:07:45.409 [TriggersWorkerPool-3-1] ERROR com.firebase.server.client.GCFEmulatorPublisher - rtdb failed to publish trigger to functions emulator with exception
com.firebase.server.client.ChannelClosedException$: The underlying socket was closed and released from the connection pool
at com.firebase.server.client.ChannelClosedException$.(SingleHostChannelPool.scala)
at com.firebase.server.client.NettyHttpBesieger.$anonfun$channelCreated$2(NettyHttpBesieger.scala:165)
at com.firebase.server.client.NettyHttpBesieger.$anonfun$channelCreated$2$adapted(NettyHttpBesieger.scala:164)
at com.firebase.server.util.package$NettyFutureAdapter.$anonfun$onComplete$1(package.scala:85)
at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:507)
at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:481)
at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:420)
at io.netty.util.concurrent.DefaultPromise.trySuccess(DefaultPromise.java:104)
at io.netty.channel.DefaultChannelPromise.trySuccess(DefaultChannelPromise.java:82)
at io.netty.channel.AbstractChannel$CloseFuture.setClosed(AbstractChannel.java:1002)
at io.netty.channel.AbstractChannel$AbstractUnsafe.doClose0(AbstractChannel.java:632)
at io.netty.channel.AbstractChannel$AbstractUnsafe.close(AbstractChannel.java:610)
at io.netty.channel.AbstractChannel$AbstractUnsafe.close(AbstractChannel.java:553)
at io.netty.channel.DefaultChannelPipeline$HeadContext.close(DefaultChannelPipeline.java:1244)
at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:618)
at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:602)
at io.netty.channel.ChannelDuplexHandler.close(ChannelDuplexHandler.java:73)
at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:618)
at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:602)
at io.netty.channel.ChannelOutboundHandlerAdapter.close(ChannelOutboundHandlerAdapter.java:71)
at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:618)
at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:602)
at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:459)
at com.firebase.server.client.PipelinedResponseReader.channelRead0(NettyHttpBesieger.scala:292)

```