go/runtime/client: Recheck failed blocks and implement max tx age

go/runtime/client/client.go

@@ -118,6 +118,10 @@ func (c *runtimeClient) SubmitTx(ctx context.Context, request *api.SubmitTxReque
 				return nil, fmt.Errorf("client: block watch channel closed unexpectedly (unknown error)")
 			}
 
+			if resp.err != nil {
+				return nil, fmt.Errorf("client: transaction failed: %w", err)
+			}
+
 			// The main event is getting a response from the watcher, handled below. If there is
 			// no result yet, this means that we need to retry publish.
 			if resp.result == nil {

go/runtime/client/watcher.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/oasisprotocol/oasis-core/go/common"
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
@@ -13,13 +14,24 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/worker/common/p2p"
 )
 
+const (
+	// Number of blocks after witch a transaction will be considered failed.
+	// XXX: this should probably be configurable.
+	maxTransactionAge = 1500
+)
+
+var errTransactionExpired = fmt.Errorf("transaction expired")
+
 type watchRequest struct {
 	id     hash.Hash
 	ctx    context.Context
 	respCh chan *watchResult
+	height int64
 }
 
-func (w *watchRequest) send(res *watchResult) error {
+func (w *watchRequest) send(res *watchResult, height int64) error {
+	w.height = height
+
 	select {
 	case <-w.ctx.Done():
 		return w.ctx.Err()
@@ -29,6 +41,7 @@ func (w *watchRequest) send(res *watchResult) error {
 }
 
 type watchResult struct {
+	err          error
 	result       []byte
 	groupVersion int64
 }
@@ -42,12 +55,14 @@ type blockWatcher struct {
 	watched map[hash.Hash]*watchRequest
 	newCh   chan *watchRequest
 
+	toBeChecked []*block.Block
+
 	stopCh chan struct{}
 }
 
-func (w *blockWatcher) checkBlock(blk *block.Block) {
+func (w *blockWatcher) checkBlock(blk *block.Block) error {
 	if blk.Header.IORoot.IsEmpty() {
-		return
+		return nil
 	}
 
 	ctx := w.common.ctx
@@ -68,8 +83,7 @@ func (w *blockWatcher) checkBlock(blk *block.Block) {
 
 	matches, err := tree.GetTransactionMultiple(ctx, txHashes)
 	if err != nil {
-		w.Logger.Error("can't get block I/O from storage", "err", err)
-		return
+		return fmt.Errorf("error getting block I/O from storage: %w", err)
 	}
 
 	for txHash, tx := range matches {
@@ -79,10 +93,12 @@ func (w *blockWatcher) checkBlock(blk *block.Block) {
 		}
 
 		// Ignore errors, the watch is getting deleted anyway.
-		_ = watch.send(res)
+		_ = watch.send(res, 0)
 		close(watch.respCh)
 		delete(w.watched, txHash)
 	}
+
+	return nil
 }
 
 func (w *blockWatcher) watch() {
@@ -103,17 +119,40 @@ func (w *blockWatcher) watch() {
 	}
 	defer blocksSub.Close()
 
+	consensusBlocks, consensusBlocksSub, err := w.common.consensus.WatchBlocks(w.common.ctx)
+	if err != nil {
+		w.Logger.Error("failed to subscribe to consensus blocks",
+			"err", err,
+		)
+		return
+	}
+	defer consensusBlocksSub.Close()
+
 	// If we were just started, refresh the committee information from any
 	// block, otherwise just from epoch transition blocks.
 	var gotInitialCommittee bool
 	// latestGroupVersion contains the latest known committee group version.
 	var latestGroupVersion int64
+	// latestHeight contains the latest known consensus block height.
+	var latestHeight int64
+
 	for {
 		// Wait for stuff to happen.
 		select {
 		case blk := <-blocks:
-			// Check block.
-			w.checkBlock(blk.Block)
+			w.toBeChecked = append(w.toBeChecked, blk.Block)
+
+			var failedBlocks []*block.Block
+			for _, b := range w.toBeChecked {
+				if err = w.checkBlock(b); err != nil {
+					w.Logger.Error("error checking block",
+						"err", err,
+						"block", b,
+					)
+					failedBlocks = append(failedBlocks, b)
+				}
+			}
+			w.toBeChecked = failedBlocks
 
 			// If this is the initial block or an epoch transition block,
 			// update latest known group version and resend all transactions.
@@ -148,19 +187,46 @@ func (w *blockWatcher) watch() {
 				res := &watchResult{
 					groupVersion: latestGroupVersion,
 				}
-				if watch.send(res) != nil {
+				if watch.send(res, latestHeight) != nil {
 					delete(w.watched, key)
 				}
 			}
 			gotInitialCommittee = true
+		case blk := <-consensusBlocks:
+			if blk == nil {
+				break
+			}
+			latestHeight = blk.Height
 
+			// Check if any transaction is due to be dropped.
+			for key, watch := range w.watched {
+				if watch.height == 0 {
+					continue
+				}
+				if (latestHeight - maxTransactionAge) < watch.height {
+					continue
+				}
+				w.Logger.Debug("dropping transaction",
+					"key", key,
+					"latest_height", latestHeight,
+					"max_transaction_age", maxTransactionAge,
+					"watch_height", watch.height,
+				)
+				res := &watchResult{
+					err: errTransactionExpired,
+				}
+				// Ignore errors, the watch is getting deleted anyway.
+				_ = watch.send(res, 0)
+				close(watch.respCh)
+				delete(w.watched, key)
+			}
 		case newWatch := <-w.newCh:
 			w.watched[newWatch.id] = newWatch
 
 			res := &watchResult{
 				groupVersion: latestGroupVersion,
 			}
-			if newWatch.send(res) != nil {
+			if newWatch.send(res, latestHeight) != nil {
 				delete(w.watched, newWatch.id)
 			}