Fix flaky shutdown

[ntkme]

This PR fixes a few shutdown reliability issues including the following flaky test on GitHub Actions windows runner:

❌ test\embedded\importer_test.dart: emits a protocol error for a response without a corresponding request ID (failed)
  
  Process `dart_sass_embedded` was killed with SIGKILL in a tear-down. Output:
      canonicalizeRequest: {
        id: 0
        importerId: 1
        url: other
        fromImport: true
      }
      
  [e] Host caused params error: Response ID 1 doesn't match any outstanding requests in compilation 4321.
      error: {
        type: PARAMS
        id: 4294967295
        message: Response ID 1 doesn't match any outstanding requests in compilation 4321.
      }
  TimeoutException after 0:00:30.000000: Test timed out after 30 seconds. See https://pub.dev/packages/test#timeouts
  dart:isolate  _RawReceivePort._handleMessage

[nex3]

Why is it necessary to pull this logic out here, rather than keeping it in ReusableIsolate?

[ntkme]

From previous implementation:

        // Parse out the compilation ID and surface the [ProtocolError] as an
        // error. This allows the [IsolateDispatcher] to notice that an error
        // has occurred and close out the underlying channel.

        // Protocol errors have already been through [_handleError] in the child
        // isolate, so we just send them as-is and close out the underlying
        // channel.

First, the motivation to remove the intermediate StreamController is to eliminate two unnecessary performance overheads:

1. To eliminate the overhead of deserializing ProtocolError packet, passing it through addError, and re-serializing it back to packet. - We don't actually care about the ProtocolError object, all we care is that the category 2 indicate the packet contains a ProtocolError.
2. To eliminate the overhead of allocating new StreamController for every single compilation.

In regarding of why the receivePort.listen is moved to IsolateDispatcher, IMO logically how the message is sent across Isolates is a contract between IsolateDispatcher and CompilationDispatcher, ReusableIsolate should purely be a passthrough helper just to help with creating and managing isolate for repeated use.

[nex3]

I think you're right that putting the decoding logic here makes the most sense. But I'm not a big fan of giving the caller responsibility for maintaining a single stream subscription across multiple borrows, and I'm not very worried about the overhead of an extra StreamController. Can we make .borrow() return a subscription that just forwards a view of the Stream<object?> which closes automatically once .return() is called?

[ntkme]

Not precisely a new Stream each time, but I made .borrow take a new onData function each time, which behaves somewhat like the previous implementation but without creating new StreamController.

[nex3]

I think this could still use some detail on why exiting immediately rather than waiting for futures to wind down is valuable. Think of it from the perspective of someone reading this without any context on the flakiness that used to occur.

[nex3]

I think you're right that putting the decoding logic here makes the most sense. But I'm not a big fan of giving the caller responsibility for maintaining a single stream subscription across multiple borrows, and I'm not very worried about the overhead of an extra StreamController. Can we make .borrow() return a subscription that just forwards a view of the Stream<object?> which closes automatically once .return() is called?

[nex3]

Thanks for working on this and helping reduce flakiness!