feat(cosmos): Chunked bundle installation

golang/cosmos/proto/agoric/swingset/msgs.proto

@@ -5,13 +5,16 @@ import "amino/amino.proto";
 import "cosmos/msg/v1/msg.proto";
 import "cosmos_proto/cosmos.proto";
 import "gogoproto/gogo.proto";
+import "agoric/swingset/swingset.proto";
 
 option go_package = "github.com/Agoric/agoric-sdk/golang/cosmos/x/swingset/types";
 
 // Transactions.
 service Msg {
   // Install a JavaScript sources bundle on the chain's SwingSet controller.
   rpc InstallBundle(MsgInstallBundle) returns (MsgInstallBundleResponse);
+  // Send a chunk of a bundle to tolerate RPC message size limits.
+  rpc SendChunk(MsgSendChunk) returns (MsgSendChunkResponse);
   // Send inbound messages.
   rpc DeliverInbound(MsgDeliverInbound) returns (MsgDeliverInboundResponse);
   // Perform a low-privilege wallet action.
@@ -127,6 +130,10 @@ message MsgProvision {
 message MsgProvisionResponse {}
 
 // MsgInstallBundle carries a signed bundle to SwingSet.
+// Of the fields bundle, compressed_bundle, and chunked_artifact, exactly one
+// must be present: bundle if complete and uncompressed, compressed_bundle if
+// complete and compressed, or chunked_artifact for a manifest of chunks to be
+// submitted in subsequent messages.
 message MsgInstallBundle {
   // Until agoric-upgrade-22 this message didn't have an amino name
   // but no clients actually used amino encoding
@@ -140,26 +147,28 @@ message MsgInstallBundle {
     (gogoproto.jsontag)  = "submitter",
     (gogoproto.moretags) = "yaml:\"submitter\""
   ];
-  // Either bundle or compressed_bundle will be set.
   // Default compression algorithm is gzip.
   bytes compressed_bundle = 3 [
     (amino.dont_omitempty) = true,
     (amino.field_name)     = "compressedBundle",
     (gogoproto.jsontag)    = "compressedBundle",
     (gogoproto.moretags)   = "yaml:\"compressedBundle\""
   ];
-  // Size in bytes of uncompression of compressed_bundle.
+  // Total size in bytes of the bundle artifact, before compression and after
+  // decompression.
   int64 uncompressed_size = 4 [
     (amino.dont_omitempty) = true,
     (amino.field_name)     = "uncompressedSize",
     (gogoproto.jsontag)    = "uncompressedSize"
   ];
+  // Declaration of a chunked bundle.
+  ChunkedArtifact chunked_artifact = 5 [
+    (amino.field_name)   = "chunkedArtifact",
+    (gogoproto.jsontag)  = "chunkedArtifact,omitempty",
+    (gogoproto.moretags) = "yaml:\"chunkedArtifact\""
+  ];
 }
 
-// MsgInstallBundleResponse is an empty acknowledgement that an install bundle
-// message has been queued for the SwingSet kernel's consideration.
-message MsgInstallBundleResponse {}
-
 // MsgCoreEval defines an SDK message for a core eval.
 message MsgCoreEval {
   option (cosmos.msg.v1.signer) = "authority";
@@ -181,3 +190,61 @@ message MsgCoreEvalResponse {
   // The result of the core eval.
   string result = 1 [(gogoproto.moretags) = "yaml:\"result\""];
 }
+
+// MsgInstallBundleResponse is either an empty acknowledgement that a bundle
+// installation message has been queued for the SwingSet kernel's
+// consideration, or for MsgInstallBundle requests that have a chunked artifact
+// manifest instead of a compressed or uncompressed bundle: the identifier
+// assigned for the chunked artifact for reference in subsequent MsgSendChunk
+// messages.
+message MsgInstallBundleResponse {
+  // The assigned identifier for a chunked artifact, if the caller is expected
+  // to call back with MsgSendChunk messages.
+  uint64 chunked_artifact_id = 1 [
+    (amino.field_name)   = "chunkedArtifactId",
+    (gogoproto.jsontag)  = "chunkedArtifactId",
+    (gogoproto.moretags) = "yaml:\"chunkedArtifactId\""
+  ];
+}
+
+// MsgSendChunk carries a chunk of an artifact through RPC to the chain.
+// Individual chunks are addressed by the chunked artifact identifier and
+// the zero-based index of the chunk among all chunks as mentioned in the
+// manifest provided to MsgInstallBundle.
+message MsgSendChunk {
+  uint64 chunked_artifact_id = 1 [
+    (amino.field_name)   = "chunkedArtifactId",
+    (gogoproto.jsontag)  = "chunkedArtifactId",
+    (gogoproto.moretags) = "yaml:\"chunkedArtifactId\""
+  ];
+  bytes submitter = 2 [
+    (amino.encoding)     = "legacy_address",
+    (amino.field_name)   = "submitter",
+    (gogoproto.casttype) = "github.com/cosmos/cosmos-sdk/types.AccAddress",
+    (gogoproto.jsontag)  = "submitter",
+    (gogoproto.moretags) = "yaml:\"submitter\""
+  ];
+  uint64 chunk_index = 3 [
+    (amino.field_name)   = "chunkIndex",
+    (gogoproto.jsontag)  = "chunkIndex",
+    (gogoproto.moretags) = "yaml:\"chunkIndex\""
+  ];
+  bytes chunk_data = 4 [
+    (amino.field_name)   = "chunkIndex",
+    (gogoproto.jsontag)  = "chunkData",
+    (gogoproto.moretags) = "yaml:\"chunkData\""
+  ];
+}
+
+// MsgSendChunkResponse is an acknowledgement that a chunk has been received by
+// the chain.
+message MsgSendChunkResponse {
+  uint64 chunked_artifact_id = 1 [
+    (amino.field_name)   = "chunkedArtifactId",
+    (gogoproto.jsontag)  = "chunkedArtifactId",
+    (gogoproto.moretags) = "yaml:\"chunkedArtifactId\""
+  ];
+  // The current state of the chunk.
+  ChunkInfo chunk = 2
+      [(amino.field_name) = "chunk", (gogoproto.jsontag) = "chunk", (gogoproto.moretags) = "yaml:\"chunk\""];
+}

golang/cosmos/proto/agoric/swingset/query.proto

@@ -23,6 +23,11 @@ service Query {
   rpc Mailbox(QueryMailboxRequest) returns (QueryMailboxResponse) {
     option (google.api.http).get = "/agoric/swingset/mailbox/{peer}";
   }
+
+  // Return the state of a pending installation.
+  rpc ChunkedArtifactStatus(QueryChunkedArtifactStatusRequest) returns (QueryChunkedArtifactStatusResponse) {
+    option (google.api.http).get = "/agoric/swingset/chunked-artifact-status/{chunked_artifact_id}";
+  }
 }
 
 // QueryParamsRequest is the request type for the Query/Params RPC method.
@@ -61,3 +66,24 @@ message QueryMailboxRequest {
 message QueryMailboxResponse {
   string value = 1 [(gogoproto.jsontag) = "value", (gogoproto.moretags) = "yaml:\"value\""];
 }
+
+// QueryChunkedArtifactStatusRequest is the request type for the Query/ChunkedArtifact RPC method.
+message QueryChunkedArtifactStatusRequest {
+  uint64 chunked_artifact_id = 1
+      [(gogoproto.jsontag) = "chunkedArtifactId", (gogoproto.moretags) = "yaml:\"chunkedArtifactId\""];
+}
+
+// QueryChunkedArtifactStatuslResponse is the response type for the Query/ChunkedArtifact RPC method.
+message QueryChunkedArtifactStatusResponse {
+  uint64 chunked_artifact_id = 1
+      [(gogoproto.jsontag) = "chunkedArtifactId", (gogoproto.moretags) = "yaml:\"chunkedArtifactId\""];
+
+  ChunkedArtifact chunked_artifact = 2
+      [(gogoproto.jsontag) = "chunkedArtifact", (gogoproto.moretags) = "yaml:\"chunkedArtifact\""];
+
+  // Start time in UNIX epoch seconds.
+  int64 start_time_unix = 3 [(gogoproto.jsontag) = "startTimeUnix", (gogoproto.moretags) = "yaml:\"startTimeUnix\""];
+
+  int64 start_block_height = 4
+      [(gogoproto.jsontag) = "startBlockHeight", (gogoproto.moretags) = "yaml:\"startBlockHeight\""];
+}

golang/cosmos/proto/agoric/swingset/swingset.proto

@@ -89,13 +89,35 @@ message Params {
   // nodes must all serialize and deserialize the existing order without
   // permuting it.
   repeated UintMapEntry vat_cleanup_budget = 6 [(gogoproto.nullable) = false];
+
+  // The maximum number of blocks that an async installation can use.  -1 is
+  // unlimited.
+  int64 installation_deadline_blocks = 7;
+
+  // The maximum number of seconds that an async installation can use.  -1 is
+  // unlimited.
+  int64 installation_deadline_seconds = 8;
+
+  // The maximum size of of a bundle (0 implies default 10MB)
+  int64 bundle_uncompressed_size_limit_bytes = 9;
+
+  // The maximum size of a bundle or artifact chunk (0 implies default 512KB)
+  int64 chunk_size_limit_bytes = 10;
 }
 
 // The current state of the module.
 message State {
   // The allowed number of items to add to queues, as determined by SwingSet.
   // Transactions which attempt to enqueue more should be rejected.
   repeated QueueSize queue_allowed = 1 [(gogoproto.nullable) = false];
+
+  // Doubly-linked list in order of start block and time.
+  uint64 first_chunked_artifact_id = 2
+      [(gogoproto.jsontag) = "first_chunked_artifact_id", (gogoproto.moretags) = "yaml:\"first_chunked_artifact_id\""];
+
+  // The last chunked artifact id that has not expired or completed.
+  uint64 last_chunked_artifact_id = 3
+      [(gogoproto.jsontag) = "lastChunkedArtifactId", (gogoproto.moretags) = "yaml:\"lastChunkedArtifactId\""];
 }
 
 // Map element of a string key to a Nat bean count.
@@ -167,3 +189,66 @@ message SwingStoreArtifact {
 
   bytes data = 2 [(gogoproto.jsontag) = "data", (gogoproto.moretags) = "yaml:\"data\""];
 }
+
+// ChunkedArtifact is the manifest for an artifact that is submitted across
+// multiple transactions, in chunks, as when using InstallBundle to submit
+// chunks.
+message ChunkedArtifact {
+  // The SHA-512 hash of the compartment-map.json file inside the bundle.
+  string sha512 = 1 [(gogoproto.jsontag) = "sha512", (gogoproto.moretags) = "yaml:\"sha512\""];
+
+  // The size of the final bundle artifact in bytes.
+  uint64 size_bytes = 2 [(gogoproto.jsontag) = "size_bytes", (gogoproto.moretags) = "yaml:\"size_bytes\""];
+
+  // Information about the chunks that will be concatenated to form this
+  // bundle.
+  repeated ChunkInfo chunks = 3 [(gogoproto.jsontag) = "chunks", (gogoproto.moretags) = "yaml:\"chunks\""];
+}
+
+// Current state of this chunk.
+enum ChunkState {
+  // Unknown state.
+  CHUNK_STATE_UNSPECIFIED = 0;
+
+  // The chunk is still in-flight.
+  CHUNK_STATE_IN_FLIGHT = 1;
+
+  // The chunk has been received.
+  CHUNK_STATE_RECEIVED = 2;
+
+  // The chunk has been processed.
+  CHUNK_STATE_PROCESSED = 3;
+};
+
+// Information about a chunk of a bundle.
+message ChunkInfo {
+  // The SHA-512 hash of the chunk contents.
+  string sha512 = 1 [(gogoproto.jsontag) = "sha512", (gogoproto.moretags) = "yaml:\"sha512\""];
+
+  // The chunk size in bytes.
+  uint64 size_bytes = 2 [(gogoproto.jsontag) = "size_bytes", (gogoproto.moretags) = "yaml:\"size_bytes\""];
+
+  // The current state of the chunk.
+  ChunkState state = 3 [(gogoproto.jsontag) = "state", (gogoproto.moretags) = "yaml:\"state\""];
+}
+
+// A node in a doubly-linked-list of chunks of a chunked artifact, as used for
+// chunked bundle installation, in order of ascending block time.
+// The keeper uses this to expediently expire stale chunks.
+message ChunkedArtifactNode {
+  // The id of the pending bundle installation.
+  uint64 chunked_artifact_id = 1
+      [(gogoproto.jsontag) = "chunkedArtifactId", (gogoproto.moretags) = "yaml:\"chunkedArtifactId\""];
+
+  // Doubly-linked list.
+  uint64 next_id = 2 [(gogoproto.jsontag) = "nextId", (gogoproto.moretags) = "yaml:\"nextId\""];
+
+  uint64 prev_id = 3 [(gogoproto.jsontag) = "prevId", (gogoproto.moretags) = "yaml:\"prevId\""];
+
+  // The time at which the pending installation began, in UNIX epoch seconds.
+  int64 start_time_unix = 4 [(gogoproto.jsontag) = "startTimeUnix", (gogoproto.moretags) = "yaml:\"startTimeUnix\""];
+
+  // The block at which the pending installation began.
+  int64 start_block_height = 5
+      [(gogoproto.jsontag) = "startBlockHeight", (gogoproto.moretags) = "yaml:\"startBlockHeight\""];
+}

golang/cosmos/x/swingset/abci.go

@@ -66,6 +66,9 @@ func EndBlock(ctx sdk.Context, keeper Keeper) ([]abci.ValidatorUpdate, error) {
 		panic(err)
 	}
 
+	// Remove expired bundle installs.
+	keeper.PruneExpiredBundleInstalls(ctx)
+
 	// Save our EndBlock status.
 	endBlockHeight = ctx.BlockHeight()
 	endBlockTime = ctx.BlockTime().Unix()

golang/cosmos/x/swingset/keeper/grpc_query.go

@@ -61,3 +61,27 @@ func (k Querier) Mailbox(c context.Context, req *types.QueryMailboxRequest) (*ty
 		Value: value,
 	}, nil
 }
+
+func (k Querier) ChunkedArtifactStatus(c context.Context, req *types.QueryChunkedArtifactStatusRequest) (*types.QueryChunkedArtifactStatusResponse, error) {
+	if req == nil {
+		return nil, status.Error(codes.InvalidArgument, "empty request")
+	}
+	ctx := sdk.UnwrapSDKContext(c)
+
+	msg := k.GetPendingBundleInstall(ctx, req.ChunkedArtifactId)
+	if msg == nil {
+		return nil, status.Error(codes.NotFound, "pending chunked artifact not found")
+	}
+
+	can := k.GetChunkedArtifactNode(ctx, req.ChunkedArtifactId)
+	if can == nil {
+		return nil, status.Error(codes.NotFound, "pending chunked artifact node not found")
+	}
+
+	return &types.QueryChunkedArtifactStatusResponse{
+		ChunkedArtifactId: req.ChunkedArtifactId,
+		ChunkedArtifact:   msg.ChunkedArtifact,
+		StartTimeUnix:     can.StartTimeUnix,
+		StartBlockHeight:  can.StartBlockHeight,
+	}, nil
+}