(unfinished) transactions in incoming messages

Consensus-layer support for runtime transactions in roothash incoming
message.

go/worker/compute/executor/committee/transactions.go

@@ -9,6 +9,8 @@ import (
 
 	cmnBackoff "github.com/oasisprotocol/oasis-core/go/common/backoff"
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
+	consensus "github.com/oasisprotocol/oasis-core/go/consensus/api"
+	roothash "github.com/oasisprotocol/oasis-core/go/roothash/api"
 	"github.com/oasisprotocol/oasis-core/go/runtime/transaction"
 	"github.com/oasisprotocol/oasis-core/go/runtime/txpool"
 	"github.com/oasisprotocol/oasis-core/go/worker/common/p2p/txsync"
@@ -39,7 +41,7 @@ func (n *Node) resolveBatchLocked(batch *unresolvedBatch, missingState NodeState
 		}
 		var ctx context.Context
 		ctx, n.missingTxsCancel = context.WithCancel(n.roundCtx)
-		go n.requestMissingTransactions(ctx)
+		go n.requestMissingTransactions(ctx, n.commonNode.CurrentConsensusBlock)
 	}
 	return resolvedBatch, nil
 }
@@ -81,7 +83,32 @@ func (n *Node) handleNewCheckedTransactions(txs []*txpool.PendingCheckTransactio
 	}
 }
 
-func (n *Node) requestMissingTransactions(ctx context.Context) {
+func (n *Node) requestMissingTransactions(ctx context.Context, consensusBlk *consensus.LightBlock) {
+	// Load transactions from roothash incoming messages.
+	func() {
+		inMsgs, err := n.commonNode.Consensus.RootHash().GetIncomingMessageQueue(ctx, &roothash.InMessageQueueRequest{
+			RuntimeID: n.commonNode.Runtime.ID(),
+			Height:    consensusBlk.Height,
+		})
+		if err != nil {
+			n.logger.Error("failed to fetch incoming runtime message queue transactions",
+				"err", err,
+			)
+			// todo: propagate sanely
+			panic(err)
+		}
+		var inMsgTxs [][]byte
+		for _, msg := range inMsgs {
+			// todo: parse msg.Data
+			if len(msg.Data) != 0 {
+				inMsgTxs = append(inMsgTxs, msg.Data)
+			}
+		}
+		if len(inMsgTxs) != 0 {
+			n.commonNode.TxPool.SubmitProposedBatch(inMsgTxs)
+		}
+	}()
+
 	requestOp := func() error {
 		// Determine what transactions are missing.
 		txHashes := func() []hash.Hash {