chore(transactor): Allow async forcing tx requests (#119)

baseapp/runners.go

@@ -164,7 +164,7 @@ func (jm *JobManager) retryableHeaderSubscriber(
 		// Handle error while subscribing.
 		if shouldRetry = (err != nil); shouldRetry {
 			jm.Logger(ctx).Error(
-				"error subscribing to filter logs, retrying...",
+				"error subscribing to block headers, retrying...",
 				"job", blockHeaderJob.RegistryKey(), "err", err,
 			)
 			return shouldRetry

core/transactor/transactor.go

@@ -159,16 +159,28 @@ func (t *TxrV2) SendTxRequest(txReq *types.Request) (string, error) {
 // ForceTxRequest immediately (whenever the sender is free from any previous sends) builds and
 // sends the tx request to the chain, after validating it.
 // NOTE: this bypasses the queue and batching even if configured to do so.
-func (t *TxrV2) ForceTxRequest(ctx context.Context, txReq *types.Request) (string, error) {
+func (t *TxrV2) ForceTxRequest(ctx context.Context, txReq *types.Request, async bool) (string, error) {
 	if err := txReq.Validate(); err != nil {
 		return "", err
 	}
 
-	go t.fire(
-		ctx,
-		&tracker.Response{MsgIDs: []string{txReq.MsgID}, InitialTimes: []time.Time{txReq.Time()}},
-		true, txReq.CallMsg,
-	)
+	if async {
+		go t.fire(
+			ctx,
+			&tracker.Response{
+				MsgIDs: []string{txReq.MsgID}, InitialTimes: []time.Time{txReq.Time()},
+			},
+			true, txReq.CallMsg,
+		)
+	} else {
+		t.fire(
+			ctx,
+			&tracker.Response{
+				MsgIDs: []string{txReq.MsgID}, InitialTimes: []time.Time{txReq.Time()},
+			},
+			true, txReq.CallMsg,
+		)
+	}
 	return txReq.MsgID, nil
 }
 

x/jobs/block_header_job.go

@@ -40,8 +40,6 @@ func (w *BlockHeaderWatcher) Subscribe(
 		return nil, nil, err
 	}
 	w.sub = sub
-
-	sCtx.Logger().Info("Subscribed to new block headers")
 	return sub, headerCh, nil
 }