fix: async CAR put() & close() for backpressure & error handling

[rvagg]

Ref: #73 /cc @jimpick

Bear with me ... This is because of the challenges of using asynciterables as streams, but we can do this!

In the current form, the promises for writer.put() and writer.close() get lost, so (a) we have no means of backpressure so if we had a fast block store + a large number of blocks and a slow reader for the output then we'd use up a lot of memory, and (b) any errors are lost to the wind.

The CAR internals for {writer,out} ("IteratorChannel") are careful to maintain backpressure between them if you choose to use await on your writer.put() (because it's not strictly necessary), so the approach in this PR should maintain a steady stream state with not excessive bytes accumulating if you have a slow writer on the other end (it would be interesting to test this assertion, we're getting into complicated territory here).

So, we set up 2 parallel activities: (1) writing blocks from the blockstore to the CAR writer and (2) reading blocks from the CAR output channel. We want to give the user the results of (2) and we don't want (1) to start until the user starts reading and we don't want blocks to be written until the bytes for previous blocks have been consumed. CAR internals take care of that latter concern as long as we can await on our writer.put() calls and waiting until the user calls [Symbol.asyncIterator]() means we can defer the blockstore read->CAR write until they start iterating. We also get to hang on to the promise where we do all the writing and by awaiting that at the last we get to propagate errors.

[vasco-santos]

LGTM! Thanks ❤️
I tried this with several large files and everything is working as expected

[alanshaw]

Wow, this cool! A bit hard to follow, but it looks like it works. I would like to give CarWriter an blocks (async) iterable so make this simple:

const out = await CarWriter.fromIterable([root], blockstore.blocks())

The writer would auto-close when the iterable is done and deal with backpressure.