Introduce 'actions' which nest under sub-status entries, including additional information pertaining to each action

Signed-off-by: Matthew Whitehead <[email protected]>

Author: matthew1001

config.md

@@ -241,7 +241,9 @@
 
 |Key|Description|Type|Default Value|
 |---|-----------|----|-------------|
-|errorHistoryCount|The number of historical errors to retain in the operation|`int`|`25`
+|maxHistoryActions|The number of actions to store per historical status updates|`int`|`50`
+|maxHistoryCount|The number of historical status updates to retain in the operation|`int`|`50`
+|maxHistorySummaryCount|The number of historical status summary records to retain in the operation|`int`|`50`
 |maxInFlight|The maximum number of transactions to have in-flight with the policy engine / blockchain transaction pool|`int`|`100`
 |nonceStateTimeout|How old the most recently submitted transaction record in our local state needs to be, before we make a request to the node to query the next nonce for a signing address|[`time.Duration`](https://pkg.go.dev/time#Duration)|`1h`
 

internal/tmconfig/tmconfig.go

@@ -30,7 +30,9 @@ var (
 	ConfirmationsBlockQueueLength                 = ffc("confirmations.blockQueueLength")
 	ConfirmationsStaleReceiptTimeout              = ffc("confirmations.staleReceiptTimeout")
 	ConfirmationsNotificationQueueLength          = ffc("confirmations.notificationQueueLength")
-	TransactionsErrorHistoryCount                 = ffc("transactions.errorHistoryCount")
+	TransactionsMaxHistoryCount                   = ffc("transactions.maxHistoryCount")
+	TransactionsMaxHistorySummaryCount            = ffc("transactions.maxHistorySummaryCount")
+	TransactionsMaxHistoryActions                 = ffc("transactions.maxHistoryActions")
 	TransactionsMaxInFlight                       = ffc("transactions.maxInFlight")
 	TransactionsNonceStateTimeout                 = ffc("transactions.nonceStateTimeout")
 	PolicyLoopInterval                            = ffc("policyloop.interval")
@@ -74,7 +76,9 @@ var MetricsConfig config.Section
 
 func setDefaults() {
 	viper.SetDefault(string(TransactionsMaxInFlight), 100)
-	viper.SetDefault(string(TransactionsErrorHistoryCount), 25)
+	viper.SetDefault(string(TransactionsMaxHistoryCount), 50)
+	viper.SetDefault(string(TransactionsMaxHistorySummaryCount), 50)
+	viper.SetDefault(string(TransactionsMaxHistoryActions), 50)
 	viper.SetDefault(string(TransactionsNonceStateTimeout), "1h")
 	viper.SetDefault(string(ConfirmationsRequired), 20)
 	viper.SetDefault(string(ConfirmationsBlockQueueLength), 50)

internal/tmmsgs/en_config_descriptions.go

@@ -45,9 +45,11 @@ var (
 	ConfigConfirmationsRequired                 = ffc("config.confirmations.required", "Number of confirmations required to consider a transaction/event final", i18n.IntType)
 	ConfigConfirmationsStaleReceiptTimeout      = ffc("config.confirmations.staleReceiptTimeout", "Duration after which to force a receipt check for a pending transaction", i18n.TimeDurationType)
 
-	ConfigTransactionsErrorHistoryCount = ffc("config.transactions.errorHistoryCount", "The number of historical errors to retain in the operation", i18n.IntType)
-	ConfigTransactionsMaxInflight       = ffc("config.transactions.maxInFlight", "The maximum number of transactions to have in-flight with the policy engine / blockchain transaction pool", i18n.IntType)
-	ConfigTransactionsNonceStateTimeout = ffc("config.transactions.nonceStateTimeout", "How old the most recently submitted transaction record in our local state needs to be, before we make a request to the node to query the next nonce for a signing address", i18n.TimeDurationType)
+	ConfigTransactionsMaxHistoryCount        = ffc("config.transactions.maxHistoryCount", "The number of historical status updates to retain in the operation", i18n.IntType)
+	ConfigTransactionsMaxHistorySummaryCount = ffc("config.transactions.maxHistorySummaryCount", "The number of historical status summary records to retain in the operation", i18n.IntType)
+	ConfigTransactionsMaxHistoryActions      = ffc("config.transactions.maxHistoryActions", "The number of actions to store per historical status updates", i18n.IntType)
+	ConfigTransactionsMaxInflight            = ffc("config.transactions.maxInFlight", "The maximum number of transactions to have in-flight with the policy engine / blockchain transaction pool", i18n.IntType)
+	ConfigTransactionsNonceStateTimeout      = ffc("config.transactions.nonceStateTimeout", "How old the most recently submitted transaction record in our local state needs to be, before we make a request to the node to query the next nonce for a signing address", i18n.TimeDurationType)
 
 	ConfigPolicyEngineName = ffc("config.policyengine.name", "The name of the policy engine to use", i18n.StringType)
 

pkg/apitypes/managed_tx.go

@@ -18,6 +18,7 @@ package apitypes
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
@@ -38,55 +39,63 @@ const (
 	TxStatusFailed TxStatus = "Failed"
 )
 
-type ManagedTXUpdate struct {
-	Time           *fftypes.FFTime    `json:"time"`
-	LastOccurrence *fftypes.FFTime    `json:"lastOccurrence"`
-	Count          int                `json:"count"`
-	Info           string             `json:"info,omitempty"`
-	Error          string             `json:"error,omitempty"`
-	MappedReason   ffcapi.ErrorReason `json:"reason,omitempty"`
-}
-
 // TxSubStatus is an intermediate status a transaction may go through
 type TxSubStatus string
 
 const (
-	// TxSubStatusRetrievingGasPrice indicates the operation is getting a gas price
-	TxSubStatusRetrievingGasPrice TxSubStatus = "RetrievingGasPrice"
-	// TxSubStatusRetrievedGasPrice indicates the operation has had gas price calculated for it
-	TxSubStatusRetrievedGasPrice TxSubStatus = "RetrievedGasPrice"
-	// TxSubStatusSubmitting indicates that the transaction is about to be submitted
-	TxSubStatusSubmitting TxSubStatus = "Submitting"
-	// TxSubStatusSubmitted indicates that the transaction has been submitted to the JSON/RPC endpoint
-	TxSubStatusSubmitted TxSubStatus = "Submitted"
-	// TxSubStatusReceivedReceipt indicates that we have received a receipt for the transaction
-	TxSubStatusReceivedReceipt TxSubStatus = "ReceivedReceipt"
-	// TxSubStatusConfirmed indicates that we have met the required number of confirmations for the transaction
+	// TxSubStatusReceived indicates the transaction has been received by the connector
+	TxSubStatusReceived TxSubStatus = "Received"
+	// TxSubStatusStale indicates the transaction is now in stale
+	TxSubStatusStale TxSubStatus = "Stale"
+	// TxSubStatusTracking indicates we are tracking progress of the transaction
+	TxSubStatusTracking TxSubStatus = "Tracking"
+	// TxSubStatusConfirmed indicates we have confirmed that the transaction has been fully processed
 	TxSubStatusConfirmed TxSubStatus = "Confirmed"
+	// TxSubStatusFailed indicates we have failed to process the transaction and it will no longer be tracked
+	TxSubStatusFailed TxSubStatus = "Failed"
 )
 
-type TxSubStatusEntry struct {
-	Time           *fftypes.FFTime `json:"time"`
-	LastOccurrence *fftypes.FFTime `json:"lastOccurrence"`
-	Status         TxSubStatus     `json:"subStatus"`
-	Count          int             `json:"count"`
+type TxHistoryRecord struct {
+	Time    *fftypes.FFTime  `json:"time"`
+	Status  TxSubStatus      `json:"subStatus"`
+	Actions []*TxActionEntry `json:"actions"`
+	Info    string           `json:"info,omitempty"`
+	Error   string           `json:"error,omitempty"`
 }
 
-// MsgString is assured to be the same, as long as the type/message is the same.
-// Does not change if the count/times are different - so allows comparison.
-func (mtu *ManagedTXUpdate) MsgString() string {
-	if mtu == nil {
-		return ""
-	}
-	msg := ""
-	if mtu.Error != "" {
-		msg = fmt.Sprintf("error: %s, ", mtu.Error)
-	}
-	if mtu.MappedReason != "" {
-		msg += fmt.Sprintf("reason: %s, ", mtu.MappedReason)
-	}
-	msg += fmt.Sprintf("info: %s", mtu.Info)
-	return msg
+type TxHistorySummaryRecord struct {
+	FirstOccurrence *fftypes.FFTime `json:"firstOccurence"`
+	Status          TxSubStatus     `json:"subStatus"`
+	Count           int             `json:"count"`
+}
+
+// TxAction is an action taken while attempting to progress a transaction between sub-states
+type TxAction string
+
+const (
+	// TxActionAssignNonce indicates that a nonce has been assigned to the transaction
+	TxActionAssignNonce TxAction = "AssignNonce"
+	// TxActionRetrieveGasPrice indicates the operation is getting a gas price
+	TxActionRetrieveGasPrice TxAction = "RetrieveGasPrice"
+	// TxActionTimeout indicates that the transaction has timed out may need intervention to progress it
+	TxActionTimeout TxAction = "Timeout"
+	// TxActionSubmitTransaction indicates that the transaction has been submitted
+	TxActionSubmitTransaction TxAction = "SubmitTransaction"
+	// TxActionReceiveReceipt indicates that we have received a receipt for the transaction
+	TxActionReceiveReceipt TxAction = "ReceiveReceipt"
+	// TxActionConfirmTransaction indicates that the transaction has been confirmed
+	TxActionConfirmTransaction TxAction = "Confirm"
+)
+
+// An action taken in order to progress a transaction, e.g. retrieve gas price from an oracle
+type TxActionEntry struct {
+	Time           *fftypes.FFTime  `json:"time"`
+	Action         TxAction         `json:"action"`
+	LastOccurrence *fftypes.FFTime  `json:"lastOccurrence"`
+	Count          int              `json:"count"`
+	LastError      *fftypes.JSONAny `json:"lastError,omitempty"`
+	LastErrorTime  *fftypes.FFTime  `json:"lastErrorTime,omitempty"`
+	LastInfo       *fftypes.JSONAny `json:"lastInfo,omitempty"`
 }
 
 // ManagedTX is the structure stored for each new transaction request, using the external ID of the operation
@@ -124,8 +133,8 @@ type ManagedTX struct {
 	LastSubmit         *fftypes.FFTime                    `json:"lastSubmit,omitempty"`
 	Receipt            *ffcapi.TransactionReceiptResponse `json:"receipt,omitempty"`
 	ErrorMessage       string                             `json:"errorMessage,omitempty"`
-	History            []*ManagedTXUpdate                 `json:"history"`
-	SubStatusHistory   []*TxSubStatusEntry                `json:"subStatusHistory"`
+	History            []*TxHistoryRecord                 `json:"history,omitempty"`
+	HistorySummary     []*TxHistorySummaryRecord          `json:"historySummary,omitempty"`
 	Confirmations      []confirmations.BlockInfo          `json:"confirmations,omitempty"`
 }
 
@@ -154,36 +163,134 @@ type TransactionUpdateReply struct {
 	TransactionHash string       `json:"transactionHash,omitempty"`
 }
 
+func (mtx *ManagedTX) CurrentSubStatus(ctx context.Context) *TxHistoryRecord {
+	if len(mtx.History) > 0 {
+		return mtx.History[len(mtx.History)-1]
+	}
+	return nil
+}
+
 // Transaction sub-status entries can be added for a given transaction so a caller
 // can see discrete steps in a transaction moving to confirmation on the blockchain.
-// There only exists a single entry in the list for each unique sub-status type. If
-// a transaction goes through the same sub-status more than once (for example if it
-// has gas recalculated for it more than once) then the "Count" and "LastOccurence" fields
-// should be updated accordingly. This design allows a caller to see when the most recent
-// sub-status changes took place and if necessary, to order them by time, but prevents
-// the potential for the sub-status list to grow indefinitely.
+// For example a transaction might have a sub-status of "Stale" if a transaction has
+// been in pending state for a given period of time. In order to progress the transaction
+// while it's in a given sub-status, certain actions might be taken (such as retrieving
+// the latest gas price for the chain). See AddSubStatusAction(). Since a transaction
+// might go through many sub-status changes before being confirmed on chain the list of
+// entries is capped at the configured number and FIFO approach used to keep within that cap.
 func (mtx *ManagedTX) AddSubStatus(ctx context.Context, subStatus TxSubStatus) {
-	// See if this status exists in the list already
-	for _, entry := range mtx.SubStatusHistory {
-		if entry.Status == subStatus {
-			entry.Count++
-			entry.LastOccurrence = fftypes.Now()
+	// See if the status being added is the same as the current status. If so we won't create
+	// a new record, just increment the total count
+	if len(mtx.History) > 0 {
+		if mtx.History[len(mtx.History)-1].Status == subStatus {
 			return
 		}
+		log.L(ctx).Debugf("Entered sub-status %s", subStatus)
 	}
 
-	// Prevent crazy run-away situations by limiting the number of unique sub-status
-	// types we will keep
-	if len(mtx.SubStatusHistory) >= 50 {
-		log.L(ctx).Warn("Number of unique sub-status types reached. Some status detail may be lost.")
+	// Do we need to remove the oldest entry to make space for this one?
+	if len(mtx.History) > 50 { // TODO - get from config
+		mtx.History = mtx.History[1:]
 	} else {
-		// If this is an entirely new status add it to the list
-		newStatus := &TxSubStatusEntry{
-			Time:           fftypes.Now(),
-			LastOccurrence: fftypes.Now(),
-			Count:          1,
-			Status:         subStatus,
+		// If this is a change in status add a new record
+		newStatus := &TxHistoryRecord{
+			Time:    fftypes.Now(),
+			Status:  subStatus,
+			Actions: make([]*TxActionEntry, 0),
+		}
+		mtx.History = append(mtx.History, newStatus)
+
+		// As was as detailed sub-status records (which might be a long list and early entries
+		// get purged at some point) we keep a separate list of all the discrete types of sub-status
+		// we've ever seen for this transaction along with a count of them. This means an early sub-status
+		// (e.g. "queued") followed by 100s of different sub-status types will still be recorded
+		newHistorySummary := true
+		for _, statusType := range mtx.HistorySummary {
+			if statusType.Status == subStatus {
+				// Just increment the counter
+				statusType.Count++
+				newHistorySummary = false
+				break
+			}
+		}
+
+		if newHistorySummary {
+			if len(mtx.HistorySummary) < 50 { // TODO - get from config
+				mtx.HistorySummary = append(mtx.HistorySummary, &TxHistorySummaryRecord{Status: subStatus, Count: 1, FirstOccurrence: fftypes.Now()})
+			} else {
+				log.L(ctx).Warnf("Reached maximum number of history summary records. New summary status will be not be recorded.")
+			}
+		}
+	}
+}
+
+// Make sure the provided value can be serialised to JSON
+func ensureValidJSON(value *fftypes.JSONAny) *fftypes.JSONAny {
+	if json.Valid([]byte(*value)) {
+		// Already valid
+		return value
+	}
+
+	// Convert to hex and wrap in a valid struct
+	hex := fmt.Sprintf("%x", []byte(*value))
+	return fftypes.JSONAnyPtr(`{"invalidJson":"` + hex + `"}`)
+}
+
+// When a transaction is in a given sub-status (e.g. "Stale") the blockchain connector
+// may perform certain actions to move it out of the status. For example it might
+// retrieve the current gas price for the chain. TxAction's represent an action taken
+// while in a given sub-status. In order to limit the number of TxAction entries in
+// a TxSubStatusEntry each action has a count of the number of occurrences and a
+// latest timestamp to indicate when it was last executed. There is a last error field
+// which can be used to indicate the most recent error that occurred, for example an
+// HTTP 4xx return code from a gas oracle. There is also an information field to record
+// arbitrary data about the action, for example the gas price retrieved from an oracle.
+func (mtx *ManagedTX) AddSubStatusAction(ctx context.Context, action TxAction, info *fftypes.JSONAny, error *fftypes.JSONAny) {
+	// An action always exists within a sub-status. If a sub-status hasn't been recorded yet we don't record the action
+	if len(mtx.History) > 0 {
+
+		// See if this action exists in the list already since we only want to update the single entry, not
+		// add a new one
+		currentSubStatus := mtx.History[len(mtx.History)-1]
+		for _, entry := range currentSubStatus.Actions {
+			if entry.Action == action {
+				entry.Count++
+				entry.LastOccurrence = fftypes.Now()
+
+				if error != nil {
+					entry.LastError = ensureValidJSON(error)
+					entry.LastErrorTime = fftypes.Now()
+				}
+
+				if info != nil {
+					entry.LastInfo = ensureValidJSON(info)
+				}
+				return
+			}
+		}
+
+		// This action hasn't been recorded yet in this sub-status. Add a new entry for it.
+		if len(currentSubStatus.Actions) >= 50 { // TODO - get from config
+			log.L(ctx).Warn("Number of unique sub-status actions. New action detail will not be recorded.")
+		} else {
+			// If this is an entirely new status add it to the list
+			newAction := &TxActionEntry{
+				Time:           fftypes.Now(),
+				Action:         action,
+				LastOccurrence: fftypes.Now(),
+				Count:          1,
+			}
+
+			if error != nil {
+				newAction.LastError = ensureValidJSON(error)
+				newAction.LastErrorTime = fftypes.Now()
+			}
+
+			if info != nil {
+				newAction.LastInfo = ensureValidJSON(info)
+			}
+
+			currentSubStatus.Actions = append(currentSubStatus.Actions, newAction)
 		}
-		mtx.SubStatusHistory = append(mtx.SubStatusHistory, newStatus)
 	}
 }