Support for uploading blobs with Put RPC call. This is the most complicated part of the service as it is stateful.

Depends on D5703. This is part 2 of 2.

I implemented it slightly differently from the C++ counterpart. I created a separate PutHandler struct to organize the code. A few notes on how it works:

• (parent diff)In each input stream message, client can send either holder, blob hash or data chunk.
• (parent diff)Put handler starts with a PutAction::None state and waits for both holder and hash to be provided (in any order, but each only once).
• (parent diff)When both of them exist, next state is determined:
  • If blob hash already exists, the AssignHolder action is scheduled and the input stream is closed by the server.
  • Otherwise, the UploadNewBlob action is initialized and handler is waiting for data chunks

Here is where this diff starts.

• Messages containng data chunks can only be sent when handler is in the UploadNewBlob state. Chunks are uploaded to S3 if big enough. The Multipart upload session is initialized when first chunk is sent.
• After input stream is closed, the finish() method is invoked, which consumes the handler and performs action depending on the enum value:
  • If no action, just return
  • For AssignHolder - just add a new row to the reverse_index table
  • In case of UploadNewBlob - finishes the S3 upload and adds rows to both DB tables

Now all gRPC handlers are implemented so running integration tests against this service is possible.
In constants set port to 50053 and add the -test suffix to both table names, run the service, then run yarn run-integration-tests blob.