refactor(client): implement tx queue to handle single tx per block limitation

Due to Akash not yet supporting multiple transactions per block, implement:
- Add transaction worker and queue system to serialize concurrent transactions
- Track block heights via node status CLI command
- Implement transaction confirmation waiting mechanism
- Add helper methods for transaction status querying
- Improve error handling for failed transactions
- Introduce new types for transaction worker and node status

Author: pcfreak30

akash/client/cli/cli.go

@@ -13,8 +13,10 @@ func envExists(key string) bool {
 }
 
 type AkashCommand struct {
-	ctx     context.Context
-	Content []string
+	client      AkashCliClient
+	ctx         context.Context
+	Content     []string
+	blockHeight string
 }
 
 type AkashCliClient interface {
@@ -29,6 +31,7 @@ func AkashCli(client AkashCliClient) AkashCommand {
 	}
 
 	return AkashCommand{
+		client:  client,
 		ctx:     client.GetContext(),
 		Content: []string{path},
 	}
@@ -38,6 +41,18 @@ func (c AkashCommand) Tx() AkashCommand {
 	return c.append("tx")
 }
 
+func (c AkashCommand) Query() AkashCommand {
+	return c.append("query")
+}
+
+func (c AkashCommand) SetHash(hash string) AkashCommand {
+	return c.append(hash)
+}
+
+func (c AkashCommand) QueryTx() AkashCommand {
+	return c.append("tx")
+}
+
 func (c AkashCommand) Deployment() AkashCommand {
 	return c.append("deployment")
 }
@@ -66,10 +81,6 @@ func (c AkashCommand) Close() AkashCommand {
 	return c.append("close")
 }
 
-func (c AkashCommand) Query() AkashCommand {
-	return c.append("query")
-}
-
 func (c AkashCommand) Market() AkashCommand {
 	return c.append("market")
 }
@@ -78,6 +89,14 @@ func (c AkashCommand) Provider() AkashCommand {
 	return c.append("provider")
 }
 
+func (c AkashCommand) Node() AkashCommand {
+	return c.append("node")
+}
+
+func (c AkashCommand) Status() AkashCommand {
+	return c.append("status")
+}
+
 func (c AkashCommand) Bid() AkashCommand {
 	return c.append("bid")
 }
@@ -206,3 +225,11 @@ func (c AkashCommand) append(str string) AkashCommand {
 	c.Content = append(c.Content, str)
 	return c
 }
+
+func (c *AkashCommand) SetBlockHeight(height string) {
+	c.blockHeight = height
+}
+
+func (c AkashCommand) GetBlockHeight() string {
+	return c.blockHeight
+}

akash/client/cli/cmd.go

@@ -10,7 +10,7 @@ import (
 	"strings"
 )
 
-func (c AkashCommand) AsCmd() *exec.Cmd {
+func (c *AkashCommand) AsCmd() *exec.Cmd {
 	return exec.Command(
 		c.Content[0],
 		c.Headless()...,
@@ -21,19 +21,18 @@ type AkashErrorResponse struct {
 	RawLog string `json:"raw_log"`
 }
 
-func (c AkashCommand) Raw() ([]byte, error) {
+func (c *AkashCommand) Raw() ([]byte, error) {
 	cmd := c.AsCmd()
-
 	tflog.Debug(c.ctx, strings.Join(cmd.Args, " "))
 
 	var errb bytes.Buffer
+
 	cmd.Stderr = &errb
+
+	// Normal handling for all other commands
 	out, err := cmd.Output()
 	if err != nil {
 		tflog.Warn(c.ctx, fmt.Sprintf("Could not execute command: %s", err.Error()))
-		if strings.Contains(errb.String(), "error unmarshalling") {
-			return c.Raw()
-		}
 
 		var akErr AkashErrorResponse
 		err := json.Unmarshal(out, &akErr)
@@ -61,14 +60,11 @@ func (c AkashCommand) DecodeJson(v any) error {
 	tflog.Debug(c.ctx, strings.Join(cmd.Args, " "))
 
 	var errb bytes.Buffer
+
 	cmd.Stderr = &errb
 	out, err := cmd.Output()
 	if err != nil {
 		tflog.Warn(c.ctx, fmt.Sprintf("Could not execute command: %s", err.Error()))
-		if strings.Contains(errb.String(), "error unmarshalling") {
-			return c.DecodeJson(v)
-		}
-
 		return errors.New(errb.String())
 	}
 

akash/client/client.go

@@ -2,12 +2,18 @@ package client
 
 import (
 	"context"
+	"errors"
+	"fmt"
+	"terraform-provider-akash/akash/client/cli"
+	"terraform-provider-akash/akash/client/types"
+	"time"
 )
 
 type AkashClient struct {
 	ctx             context.Context
 	Config          AkashProviderConfiguration
 	transactionNote string
+	txQueue         chan types.TxRequest
 }
 
 type AkashProviderConfiguration struct {
@@ -36,5 +42,72 @@ func (ak *AkashClient) SetGlobalTransactionNote(note string) {
 }
 
 func New(ctx context.Context, configuration AkashProviderConfiguration) *AkashClient {
-	return &AkashClient{ctx: ctx, Config: configuration}
+	client := &AkashClient{
+		ctx:     ctx,
+		Config:  configuration,
+		txQueue: make(chan types.TxRequest, 100),
+	}
+
+	go client.txWorker()
+
+	return client
+}
+
+func (ak *AkashClient) txWorker() {
+	for req := range ak.txQueue {
+		txId, err := req.Handler()
+		if err != nil {
+			req.Result <- types.TxResult{Err: err}
+		}
+
+		err = ak.waitForTx(txId)
+		req.Result <- types.TxResult{Err: err}
+		close(req.Result)
+	}
+}
+
+func (ak *AkashClient) waitForTx(txHash string) error {
+	timer := time.NewTimer(30 * time.Second)
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-timer.C:
+			return errors.New("timeout waiting for transaction confirmation")
+		default:
+			cmd := cli.AkashCli(ak).Query().Tx().SetHash(txHash).
+				SetNode(ak.Config.Node).
+				OutputJson()
+
+			var txResponse types.Transaction
+			if err := cmd.DecodeJson(&txResponse); err != nil {
+				time.Sleep(500 * time.Millisecond)
+				continue
+			}
+
+			if txResponse.Failed() {
+				return fmt.Errorf("transaction failed: %s", txResponse.RawLog)
+			}
+
+			return nil
+		}
+	}
+}
+
+func (ak *AkashClient) WaitForTransaction(handler types.TxHandler) error {
+	resultChan := make(chan types.TxResult, 1)
+	ak.txQueue <- types.TxRequest{
+		Handler: handler,
+		Result:  resultChan,
+	}
+
+	timer := time.NewTimer(30 * time.Second)
+	defer timer.Stop()
+
+	select {
+	case <-timer.C:
+		return errors.New("timeout waiting for transaction confirmation")
+	case result := <-resultChan:
+		return result.Err
+	}
 }

akash/client/deployment.go

@@ -60,13 +60,33 @@ func transactionCreateDeployment(ak *AkashClient, manifestLocation string) (type
 		SetNode(ak.Config.Node).
 		SetNote(ak.transactionNote).OutputJson()
 
-	transaction := types.Transaction{}
-	if err := cmd.DecodeJson(&transaction); err != nil {
+	var transaction types.Transaction
+
+	err := ak.WaitForTransaction(func() (string, error) {
+		if err := cmd.DecodeJson(&transaction); err != nil {
+			return "", err
+		}
+		return transaction.TxHash, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	if transaction.Failed() {
+		return nil, fmt.Errorf("transaction failed: %s", transaction.RawLog)
+	}
+
+	// Query the transaction to get its events
+	queryCmd := cli.AkashCli(ak).Query().QueryTx().SetHash(transaction.TxHash).
+		SetNode(ak.Config.Node).
+		OutputJson()
+
+	if err := queryCmd.DecodeJson(&transaction); err != nil {
 		return nil, err
 	}
 
 	if len(transaction.Logs) == 0 {
-		return nil, errors.New(fmt.Sprintf("something went wrong: %s", transaction.RawLog))
+		return nil, errors.New("no transaction logs found")
 	}
 
 	return transaction.Logs[0].Events[0].Attributes, nil
@@ -80,12 +100,18 @@ func (ak *AkashClient) DeleteDeployment(dseq string, owner string) error {
 		SetNode(ak.Config.Node).
 		SetNote(ak.transactionNote).AutoAccept().OutputJson()
 
-	out, err := cmd.Raw()
+	var transaction types.Transaction
+	err := ak.WaitForTransaction(func() (string, error) {
+		if err := cmd.DecodeJson(&transaction); err != nil {
+			return "", err
+		}
+		return transaction.TxHash, nil
+	})
 	if err != nil {
 		return err
 	}
 
-	tflog.Debug(ak.ctx, fmt.Sprintf("Response: %s", out))
+	tflog.Debug(ak.ctx, fmt.Sprintf("Response: %s", transaction.RawLog))
 
 	return nil
 }
@@ -99,12 +125,19 @@ func (ak *AkashClient) UpdateDeployment(dseq string, manifestLocation string) er
 		SetNote(ak.transactionNote).
 		GasAuto().SetGasAdjustment(1.5).SetGasPrices().SetSignMode("amino-json").AutoAccept().OutputJson()
 
-	out, err := cmd.Raw()
+	var transaction types.Transaction
+	err := ak.WaitForTransaction(func() (string, error) {
+		if err := cmd.DecodeJson(&transaction); err != nil {
+			return "", err
+		}
+		return transaction.TxHash, nil
+	})
+
 	if err != nil {
 		return err
 	}
 
-	tflog.Debug(ak.ctx, fmt.Sprintf("Response: %s", out))
+	tflog.Debug(ak.ctx, fmt.Sprintf("Response: %s", transaction.RawLog))
 
 	return nil
 }

akash/client/lease.go

@@ -1,7 +1,10 @@
 package client
 
 import (
+	"encoding/json"
+	"strings"
 	"terraform-provider-akash/akash/client/cli"
+	"terraform-provider-akash/akash/client/types"
 )
 
 func (ak *AkashClient) CreateLease(seqs Seqs, provider string) (string, error) {
@@ -11,8 +14,28 @@ func (ak *AkashClient) CreateLease(seqs Seqs, provider string) (string, error) {
 		DefaultGas().SetChainId(ak.Config.ChainId).SetKeyringBackend(ak.Config.KeyringBackend).
 		SetNote(ak.transactionNote).AutoAccept().SetNode(ak.Config.Node).OutputJson()
 
-	out, err := cmd.Raw()
-	if err != nil {
+	var out []byte
+	var err error
+	var transaction types.Transaction
+
+	if err = ak.WaitForTransaction(func() (string, error) {
+		out, err = cmd.Raw()
+		if err != nil {
+			return "", err
+		}
+
+		err = json.Unmarshal(out, &transaction)
+		if err != nil {
+			return "", err
+		}
+
+		err = json.NewDecoder(strings.NewReader(string(out))).Decode(&transaction)
+		if err != nil {
+			return "", err
+		}
+
+		return transaction.TxHash, nil
+	}); err != nil {
 		return "", err
 	}
 

akash/client/types/node_status.go

@@ -0,0 +1,9 @@
+package types
+
+type NodeStatus struct {
+	SyncInfo struct {
+		LatestBlockHeight string `json:"latest_block_height"`
+		LatestBlockTime   string `json:"latest_block_time"` 
+		CatchingUp        bool   `json:"catching_up"`
+	} `json:"SyncInfo"`
+}

akash/client/types/transactions.go

@@ -4,6 +4,7 @@ import "errors"
 
 type Transaction struct {
 	Height string           `json:"height"`
+	TxHash string          `json:"txhash"`
 	Logs   []TransactionLog `json:"logs"`
 	RawLog string           `json:"raw_log"`
 }
@@ -33,3 +34,23 @@ func (a TransactionEventAttributes) Get(key string) (string, error) {
 
 	return "", errors.New("attribute not found")
 }
+
+func (t Transaction) Failed() bool {
+	return len(t.Logs) == 0
+}
+
+func (t Transaction) GetEventsByType(eventType string) []TransactionEvent {
+	if len(t.Logs) == 0 {
+		return nil
+	}
+	
+	var matches []TransactionEvent
+	for _, log := range t.Logs {
+		for _, event := range log.Events {
+			if event.Type == eventType {
+				matches = append(matches, event)
+			}
+		}
+	}
+	return matches
+}

akash/client/types/tx_worker.go

@@ -0,0 +1,12 @@
+package types
+
+type TxHandler func() (string, error)
+
+type TxRequest struct {
+	Handler TxHandler
+	Result  chan TxResult
+}
+
+type TxResult struct {
+	Err error
+}