TXM In-memory: address_state methods: step 3-03

common/txmgr/address_state.go

@@ -1,6 +1,7 @@
 package txmgr
 
 import (
+	"fmt"
 	"sync"
 	"time"
 
@@ -124,12 +125,12 @@
 }
 
 // countTransactionsByState returns the number of transactions that are in the given state
 func (as *addressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) countTransactionsByState(txState txmgrtypes.TxState) int {
 	return 0
 }
 
 // findTxWithIdempotencyKey returns the transaction with the given idempotency key. If no transaction is found, nil is returned.
 func (as *addressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) findTxWithIdempotencyKey(key string) *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] {
 	return nil
 }
 
@@ -137,7 +138,7 @@
 // If txIDs are provided, only the transactions with those IDs are considered.
 // If no txIDs are provided, all transactions in the given states are considered.
 // If no txStates are provided, all transactions are considered.
 func (as *addressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) applyToTxsByState(
 	txStates []txmgrtypes.TxState,
 	fn func(*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]),
 	txIDs ...int64,
@@ -149,7 +150,7 @@
 // If no txIDs are provided, all transactions are considered.
 // If no txStates are provided, all transactions are considered.
 // The txFilter is applied to the transactions and the txAttemptFilter is applied to the attempts.
 func (as *addressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) findTxAttempts(
 	txStates []txmgrtypes.TxState,
 	txFilter func(*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) bool,
 	txAttemptFilter func(*txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) bool,
@@ -171,55 +172,145 @@
 }
 
 // pruneUnstartedTxQueue removes the transactions with the given IDs from the unstarted transaction queue.
 func (as *addressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) pruneUnstartedTxQueue(ids []int64) {
 }
 
 // deleteTxs removes the transactions with the given IDs from the address state.
 func (as *addressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) deleteTxs(txs ...txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) {
 }
 
 // peekNextUnstartedTx returns the next unstarted transaction in the queue without removing it from the unstarted queue.
 func (as *addressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) peekNextUnstartedTx() (*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], error) {
 	return nil, nil
 }
 
 // peekInProgressTx returns the in-progress transaction without removing it from the in-progress state.
 func (as *addressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) peekInProgressTx() (*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], error) {
 	return nil, nil
 }
 
 // addTxToUnstarted adds the given transaction to the unstarted queue.
 func (as *addressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) addTxToUnstarted(tx *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) error {
 	return nil
 }
 
 // moveUnstartedToInProgress moves the next unstarted transaction to the in-progress state.
+// The supplied txAttempt is added to the transaction.
+// It returns an error if there is already a transaction in progress.
+// It returns an error if there is no unstarted transaction to move to in_progress.
 func (as *addressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) moveUnstartedToInProgress(
-	etx *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE],
-	txAttempt *txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE],
+	txID int64, seq SEQ, broadcastAt time.Time, initialBroadcastAt time.Time,
+	txAttempt txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE],
 ) error {
+	as.Lock()
+	defer as.Unlock()
+
+	if as.inprogressTx != nil {
+		return fmt.Errorf("move_unstarted_to_in_progress: address %s already has a transaction in progress", as.fromAddress)
+	}
+
+	tx := as.unstartedTxs.RemoveTxByID(txID)
+	if tx == nil {
+		return fmt.Errorf("move_unstarted_to_in_progress: no unstarted transaction to move to in_progress")
+	}
+	tx.State = TxInProgress
+	tx.TxAttempts = []txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]{txAttempt}
+	tx.Sequence = &seq
+	tx.BroadcastAt = &broadcastAt
+	tx.InitialBroadcastAt = &initialBroadcastAt
+
+	as.attemptHashToTxAttempt[txAttempt.Hash] = &txAttempt
+	as.inprogressTx = tx
+
 	return nil
 }
 
 // moveConfirmedMissingReceiptToUnconfirmed moves the confirmed missing receipt transaction to the unconfirmed state.
+// It returns an error if there is no confirmed missing receipt transaction with the given ID.
 func (as *addressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) moveConfirmedMissingReceiptToUnconfirmed(
 	txID int64,
 ) error {
+	as.Lock()
+	defer as.Unlock()
+
+	tx, ok := as.confirmedMissingReceiptTxs[txID]
+	if !ok || tx == nil {
+		return fmt.Errorf("move_confirmed_missing_receipt_to_unconfirmed: no confirmed_missing_receipt transaction with ID %d", txID)
+	}
+
+	tx.State = TxUnconfirmed
+	as.unconfirmedTxs[tx.ID] = tx
+	delete(as.confirmedMissingReceiptTxs, tx.ID)
+
 	return nil
 }
 
 // moveInProgressToUnconfirmed moves the in-progress transaction to the unconfirmed state.
+// It returns an error if there is no transaction in progress.
+// It returns an error if the transaction in progress does not have an attempt with the given ID.
 func (as *addressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) moveInProgressToUnconfirmed(
-	etx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE],
-	txAttempt txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE],
+	txError null.String, broadcastAt time.Time, initialBroadcastAt time.Time,
+	txAttemptID int64,
 ) error {
+	as.Lock()
+	defer as.Unlock()
+
+	tx := as.inprogressTx
+	if tx == nil {
+		return fmt.Errorf("move_in_progress_to_unconfirmed: no transaction in progress")
+	}
+
+	found := false
+	for i := 0; i < len(tx.TxAttempts); i++ {
+		txAttempt := tx.TxAttempts[i]
+		if txAttempt.ID == txAttemptID {
+			txAttempt.State = txmgrtypes.TxAttemptBroadcast
+			tx.TxAttempts[i] = txAttempt
+			found = true
+			break
+		}
+	}
+	if !found {
+		return fmt.Errorf("move_in_progress_to_unconfirmed: transaction in progress does not have an attempt with ID %d", txAttemptID)
+	}
+
+	tx.State = TxUnconfirmed
+	tx.Error = txError
+	tx.BroadcastAt = &broadcastAt
+	tx.InitialBroadcastAt = &initialBroadcastAt
+
+	as.unconfirmedTxs[tx.ID] = tx
+	as.inprogressTx = nil
+
 	return nil
 }
 
 // moveUnconfirmedToConfirmed moves the unconfirmed transaction to the confirmed state.
+// It returns an error if there is no unconfirmed transaction with the given ID.
+// It returns an error if there is no attempt with the given receipt.
 func (as *addressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) moveUnconfirmedToConfirmed(
 	receipt txmgrtypes.ChainReceipt[TX_HASH, BLOCK_HASH],
 ) error {
+	as.Lock()
+	defer as.Unlock()
+
+	txAttempt, ok := as.attemptHashToTxAttempt[receipt.GetTxHash()]
+	if !ok {
+		return fmt.Errorf("move_unconfirmed_to_confirmed: no attempt found with receipt %v", receipt)
+	}
+	// TODO(jtw): not sure how to set blocknumber, transactionindex, and receipt on conflict
+	txAttempt.Receipts = []txmgrtypes.ChainReceipt[TX_HASH, BLOCK_HASH]{receipt}
+	tx, ok := as.unconfirmedTxs[txAttempt.TxID]
+	if !ok {
+		return fmt.Errorf("move_unconfirmed_to_confirmed: no unconfirmed transaction with ID %d", txAttempt.TxID)
+	}
+	txAttempt.State = txmgrtypes.TxAttemptBroadcast
+	if txAttempt.BroadcastBeforeBlockNum == nil {
+		blockNum := receipt.GetBlockNumber().Int64()
+		txAttempt.BroadcastBeforeBlockNum = &blockNum
+	}
+	tx.State = TxConfirmed
+
 	return nil
 }