GRPC entrypoint to download a blob (#2044)

## Motivation

We need to be able to download blobs directly from the Proxy as part of the user blobs work. Fixes #2028

## Proposal

This PR does a few things:
* Adds a storage connection to the Proxy
* Adds an entrypoint for downloading a blob given a blob ID

## Test Plan

WIP

docker/Dockerfile

@@ -110,4 +110,5 @@ COPY --from=binaries \
 COPY --chmod=755 \
     docker/server-entrypoint.sh \
     docker/server-init.sh \
+    docker/proxy-init.sh \
     ./

docker/proxy-init.sh

@@ -0,0 +1,19 @@
+#!/bin/sh
+
+while true; do
+  ./linera-db check_existence --storage "scylladb:tcp:scylla-client.scylla.svc.cluster.local:9042"
+  status=$?
+
+  if [ $status -eq 0 ]; then
+    echo "Database already exists, no need to initialize."
+    exit 0
+  else
+    # We rely on the shards to initialize the database, so just wait here
+    if [ $status -eq 1 ]; then
+        echo "Database does not exist, retrying in 5 seconds..."
+    else
+        echo "An unexpected error occurred (status: $status), retrying in 5 seconds..."
+    fi
+    sleep 5
+  fi
+done

kubernetes/linera-validator/templates/proxy.yaml

@@ -19,7 +19,6 @@ spec:
   type: NodePort
 
 ---
-
 apiVersion: v1
 kind: Service
 metadata:
@@ -36,7 +35,6 @@ spec:
   type: NodePort
 
 ---
-
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -53,6 +51,16 @@ spec:
     spec:
       serviceAccountName: linera-admin
       terminationGracePeriodSeconds: 10
+      initContainers:
+        - name: linera-proxy-initializer
+          image: {{ .Values.lineraImage }}
+          imagePullPolicy: {{ .Values.lineraImagePullPolicy }}
+          command: ["./proxy-init.sh"]
+          env:
+            - name: RUST_LOG
+              value: {{ .Values.logLevel }}
+            - name: RUST_BACKTRACE
+              value: "1"
       containers:
         - name: linera-proxy
           imagePullPolicy: {{ .Values.lineraImagePullPolicy }}
@@ -62,7 +70,14 @@ spec:
               name: linera-port
             - containerPort: 20100
               name: linera-port-int
-          command: ["./linera-proxy"]
+          command:
+            [
+              "./linera-proxy",
+              "--storage",
+              "scylladb:tcp:scylla-client.scylla.svc.cluster.local:9042",
+              "--genesis",
+              "/config/genesis.json",
+            ]
           args: ["/config/server.json"]
           env:
             - name: RUST_LOG
@@ -90,3 +105,5 @@ spec:
             items:
               - key: serverConfig
                 path: server.json
+              - key: genesisConfig
+                path: genesis.json

linera-base/src/data_types.rs

@@ -737,8 +737,9 @@ impl std::str::FromStr for OracleResponse {
 /// A blob of binary data.
 #[derive(Eq, PartialEq, Debug, Hash, Clone, Serialize, Deserialize)]
 pub struct Blob {
+    /// Bytes of the binary blob.
     #[serde(with = "serde_bytes")]
-    bytes: Vec<u8>,
+    pub bytes: Vec<u8>,
 }
 
 impl Blob {
@@ -794,6 +795,16 @@ impl HashedBlob {
         };
         blob.into_hashed()
     }
+
+    /// Returns a reference to the inner `Blob`, without the hash.
+    pub fn blob(&self) -> &Blob {
+        &self.blob
+    }
+
+    /// Moves ownership of the blob of binary data
+    pub fn into_inner(self) -> Blob {
+        self.blob
+    }
 }
 
 impl Serialize for HashedBlob {

linera-core/src/node.rs

@@ -5,7 +5,7 @@
 use futures::stream::{BoxStream, LocalBoxStream, Stream};
 use linera_base::{
     crypto::CryptoError,
-    data_types::{ArithmeticError, BlockHeight, HashedBlob},
+    data_types::{ArithmeticError, Blob, BlockHeight, HashedBlob},
     identifiers::{BlobId, ChainId},
 };
 use linera_chain::{
@@ -80,6 +80,8 @@ pub trait LocalValidatorNode {
         &mut self,
         chains: Vec<ChainId>,
     ) -> Result<Self::NotificationStream, NodeError>;
+
+    async fn download_blob(&mut self, blob_id: BlobId) -> Result<Blob, NodeError>;
 }
 
 /// Turn an address into a validator node.

linera-core/src/unit_tests/test_utils.rs

@@ -12,7 +12,7 @@ use futures::{lock::Mutex, Future};
 use linera_base::{
     crypto::*,
     data_types::*,
-    identifiers::{ChainDescription, ChainId},
+    identifiers::{BlobId, ChainDescription, ChainId},
 };
 use linera_chain::data_types::{
     BlockProposal, Certificate, HashedCertificateValue, LiteCertificate,
@@ -158,6 +158,11 @@ where
     async fn get_version_info(&mut self) -> Result<VersionInfo, NodeError> {
         Ok(Default::default())
     }
+
+    async fn download_blob(&mut self, blob_id: BlobId) -> Result<Blob, NodeError> {
+        self.spawn_and_receive(move |validator, sender| validator.do_download_blob(blob_id, sender))
+            .await
+    }
 }
 
 impl<S> LocalValidatorClient<S>
@@ -324,6 +329,21 @@ where
         let stream: NotificationStream = Box::pin(UnboundedReceiverStream::new(rx));
         sender.send(Ok(stream))
     }
+
+    async fn do_download_blob(
+        self,
+        blob_id: BlobId,
+        sender: oneshot::Sender<Result<Blob, NodeError>>,
+    ) -> Result<(), Result<Blob, NodeError>> {
+        let validator = self.client.lock().await;
+        let hashed_blob = validator
+            .state
+            .storage_client()
+            .read_hashed_blob(blob_id)
+            .await
+            .map_err(Into::into);
+        sender.send(hashed_blob.map(|hashed_blob| hashed_blob.blob().clone()))
+    }
 }
 
 #[derive(Clone)]

linera-rpc/proto/rpc.proto

@@ -51,6 +51,9 @@ service ValidatorNode {
 
   // Request the node's version info.
   rpc GetVersionInfo(google.protobuf.Empty) returns (VersionInfo);
+
+  // Downloads a blob.
+  rpc DownloadBlob(BlobId) returns (Blob);
 }
 
 // Information about the Linera crate version the validator is running
@@ -147,7 +150,7 @@ message ChainInfoQuery {
 
   // Request a signed vote for fallback mode.
   bool request_fallback = 11;
-  
+
   // Query a value that contains a binary blob (e.g. bytecode) required by this chain.
   optional bytes request_blob = 12;
 }
@@ -171,7 +174,7 @@ message BlockProposal {
 
   // A certificate for a validated block that justifies the proposal in this round.
   optional bytes validated = 6;
-  
+
   // Required blob
   bytes blobs = 7;
 }
@@ -216,7 +219,7 @@ message Certificate {
   // Wait until all outgoing cross-chain messages from this certificate have
   // been received by the target chains.
   bool wait_for_outgoing_messages = 6;
-  
+
   // Blobs required by this certificate
   bytes blobs = 7;
 }
@@ -237,6 +240,16 @@ message Signature {
   bytes bytes = 1;
 }
 
+// A content-addressed blob ID i.e. the hash of the Blob.
+message BlobId {
+  bytes bytes = 1;
+}
+
+// A blob of binary data.
+message Blob {
+  bytes bytes = 1;
+}
+
 // Response to `ChainInfoQuery`
 message ChainInfoResponse {
   // bincode-encoded chain info

linera-rpc/src/client.rs

@@ -1,7 +1,10 @@
 // Copyright (c) Zefchain Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-use linera_base::{data_types::HashedBlob, identifiers::ChainId};
+use linera_base::{
+    data_types::{Blob, HashedBlob},
+    identifiers::{BlobId, ChainId},
+};
 use linera_chain::data_types::{
     BlockProposal, Certificate, HashedCertificateValue, LiteCertificate,
 };
@@ -141,4 +144,13 @@ impl ValidatorNode for Client {
             Client::Simple(simple_client) => simple_client.get_version_info().await?,
         })
     }
+
+    async fn download_blob(&mut self, blob_id: BlobId) -> Result<Blob, NodeError> {
+        Ok(match self {
+            Client::Grpc(grpc_client) => grpc_client.download_blob(blob_id).await?,
+
+            #[cfg(with_simple_network)]
+            Client::Simple(simple_client) => simple_client.download_blob(blob_id).await?,
+        })
+    }
 }

linera-rpc/src/grpc/client.rs

@@ -4,7 +4,10 @@
 use std::{iter, time::Duration};
 
 use futures::{future, stream, StreamExt};
-use linera_base::{data_types::HashedBlob, identifiers::ChainId};
+use linera_base::{
+    data_types::{Blob, HashedBlob},
+    identifiers::{BlobId, ChainId},
+};
 use linera_chain::data_types;
 #[cfg(web)]
 use linera_core::node::{
@@ -26,7 +29,8 @@ use {
 
 use super::{
     api::{
-        chain_info_result::Inner, validator_node_client::ValidatorNodeClient, SubscriptionRequest,
+        self, chain_info_result::Inner, validator_node_client::ValidatorNodeClient,
+        SubscriptionRequest,
     },
     transport, GrpcError, GRPC_MAX_MESSAGE_SIZE,
 };
@@ -260,6 +264,16 @@ impl ValidatorNode for GrpcClient {
     async fn get_version_info(&mut self) -> Result<VersionInfo, NodeError> {
         Ok(self.client.get_version_info(()).await?.into_inner().into())
     }
+
+    #[instrument(target = "grpc_client", skip_all, err, fields(address = self.address))]
+    async fn download_blob(&mut self, blob_id: BlobId) -> Result<Blob, NodeError> {
+        Ok(self
+            .client
+            .download_blob(<BlobId as Into<api::BlobId>>::into(blob_id))
+            .await?
+            .into_inner()
+            .into())
+    }
 }
 
 #[cfg(not(web))]

linera-rpc/src/grpc/conversions.rs

@@ -3,9 +3,9 @@
 
 use linera_base::{
     crypto::{CryptoError, CryptoHash, PublicKey, Signature},
-    data_types::BlockHeight,
+    data_types::{Blob, BlockHeight},
     ensure,
-    identifiers::{ChainId, Owner},
+    identifiers::{BlobId, ChainId, Owner},
 };
 use linera_chain::data_types::{
     BlockAndRound, BlockProposal, Certificate, HashedCertificateValue, LiteCertificate, LiteValue,
@@ -530,6 +530,34 @@ impl TryFrom<api::Owner> for Owner {
     }
 }
 
+impl TryFrom<api::BlobId> for BlobId {
+    type Error = GrpcProtoConversionError;
+
+    fn try_from(blob_id: api::BlobId) -> Result<Self, Self::Error> {
+        Ok(Self(CryptoHash::try_from(blob_id.bytes.as_slice())?))
+    }
+}
+
+impl From<BlobId> for api::BlobId {
+    fn from(blob_id: BlobId) -> Self {
+        Self {
+            bytes: blob_id.0.as_bytes().to_vec(),
+        }
+    }
+}
+
+impl From<Blob> for api::Blob {
+    fn from(blob: Blob) -> Self {
+        Self { bytes: blob.bytes }
+    }
+}
+
+impl From<api::Blob> for Blob {
+    fn from(blob: api::Blob) -> Self {
+        Self { bytes: blob.bytes }
+    }
+}
+
 #[cfg(test)]
 pub mod tests {
     use std::{borrow::Cow, fmt::Debug};