message

package
v1.28.3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 7, 2026 License: Apache-2.0, MIT Imports: 39 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	MessageReplaceGasBumpPercent     int64 = 120
	MessageReplaceMaxPerSenderPerRun int   = 10
)
View Source 
const (
	EthMessageReplaceGasBumpPercent     int64 = 120
	EthMessageReplaceMaxPerSenderPerRun int   = 10
)
View Source 
const MinConfidence = 5
View Source 
const (
	MinEthConfidence = 1
)
View Source 
const (
	ReplaceStuckEpochs abi.ChainEpoch = 10
)

Variables

View Source 
var EthMessageReplaceMaxFee = abi.NewTokenAmount(50_000_000_000_000_000) // 0.05 FIL
View Source 
var MessageReplaceMaxFee = abi.TokenAmount(types.MustParseFIL("0.05 FIL"))
View Source 
var NBI uint64 = 10
View Source 
var SendLockedWait = 100 * time.Millisecond

Functions

func NewMessageReplacer added in v1.28.3 

func NewMessageReplacer(ctx context.Context, cfg ReplacerConfig) error

func NewSender 

func NewSender(api SenderAPI, signer SignerAPI, db *harmonydb.DB, maximizeFeeCap bool) (*Sender, *SendTask)

NewSender creates a new Sender.

func NewSenderETH added in v1.25.1 

func NewSenderETH(client ethchain.EthClient, db *harmonydb.DB) (*SenderETH, *SendTaskETH)

NewSenderETH creates a new SenderETH.

Types

type EthClient added in v1.26.0 

type EthClient interface {
	TransactionReceipt(ctx context.Context, txHash common.Hash) (*types.Receipt, error)
	TransactionByHash(ctx context.Context, txHash common.Hash) (tx *types.Transaction, isPending bool, err error)
	HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error)
}

EthClient is an interface for the Ethereum client operations we need

type EthMessageWaitsGCTask added in v1.28.3 

type EthMessageWaitsGCTask struct {
	// contains filtered or unexported fields
}

func NewMessageWaitsEthGCTask added in v1.28.3 

func NewMessageWaitsEthGCTask(db *harmonydb.DB, eth ethMsgWaitsGC) *EthMessageWaitsGCTask

func (*EthMessageWaitsGCTask) Adder added in v1.28.3 

func (t *EthMessageWaitsGCTask) Adder(taskFunc harmonytask.AddTaskFunc)

func (*EthMessageWaitsGCTask) CanAccept added in v1.28.3 

func (t *EthMessageWaitsGCTask) CanAccept(ids []harmonytask.TaskID, engine *harmonytask.TaskEngine) ([]harmonytask.TaskID, error)

func (*EthMessageWaitsGCTask) Do added in v1.28.3 

func (t *EthMessageWaitsGCTask) Do(ctx context.Context, taskID harmonytask.TaskID, stillOwned func() bool) (done bool, err error)

func (*EthMessageWaitsGCTask) TypeDetails added in v1.28.3 

func (t *EthMessageWaitsGCTask) TypeDetails() harmonytask.TaskTypeDetails

type EthReplacerConfig added in v1.28.3 

type EthReplacerConfig struct {
	Client ethchain.EthClient
}

type EthTransactionManager added in v1.26.0 

type EthTransactionManager interface {
	// AssignPendingToMachine assigns pending transactions to a machine for processing
	AssignPendingToMachine(ctx context.Context, machineID int64) (int, error)

	// GetPendingForMachine gets all pending transactions assigned to a machine
	GetPendingForMachine(ctx context.Context, machineID int64) ([]PendingEthTx, error)

	// UpdateToConfirmed updates a transaction to confirmed status with all the details
	UpdateToConfirmed(ctx context.Context, signedTxHash string, blockNumber int64, confirmedTxHash string, txData []byte, receipt []byte, success bool) error
}

EthTransactionManager provides a simple interface for managing Ethereum transactions in the database

type FilecoinReplacerConfig added in v1.28.3 

type FilecoinReplacerConfig struct {
	API    MessageReplaceAPI
	Signer SignerAPI
}

type HarmonyEthTxManager added in v1.26.0 

type HarmonyEthTxManager struct {
	// contains filtered or unexported fields
}

HarmonyEthTxManager is the real implementation using HarmonyDB

func NewHarmonyEthTxManager added in v1.26.0 

func NewHarmonyEthTxManager(db *harmonydb.DB) *HarmonyEthTxManager

NewHarmonyEthTxManager creates a new HarmonyEthTxManager

func (*HarmonyEthTxManager) AssignPendingToMachine added in v1.26.0 

func (h *HarmonyEthTxManager) AssignPendingToMachine(ctx context.Context, machineID int64) (int, error)

AssignPendingToMachine assigns pending transactions to a machine for processing

func (*HarmonyEthTxManager) GetPendingForMachine added in v1.26.0 

func (h *HarmonyEthTxManager) GetPendingForMachine(ctx context.Context, machineID int64) ([]PendingEthTx, error)

GetPendingForMachine gets all pending transactions assigned to a machine

func (*HarmonyEthTxManager) UpdateToConfirmed added in v1.26.0 

func (h *HarmonyEthTxManager) UpdateToConfirmed(ctx context.Context, signedTxHash string, blockNumber int64, confirmedTxHash string, txData []byte, receipt []byte, success bool) error

UpdateToConfirmed updates a transaction to confirmed status with all the details

type MessageReplaceAPI added in v1.28.3 

type MessageReplaceAPI interface {
	StateMinerInfo(ctx context.Context, actor address.Address, tsk types.TipSetKey) (api.MinerInfo, error)
	StateAccountKey(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error)
	StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error)
	GasEstimateMessageGas(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec, tsk types.TipSetKey) (*types.Message, error)
	MpoolPush(ctx context.Context, smsg *types.SignedMessage) (cid.Cid, error)
	WalletBalance(ctx context.Context, addr address.Address) (big.Int, error)
}

type MessageWaiterApi 

type MessageWaiterApi interface {
	StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error)
	ChainGetTipSetByHeight(context.Context, abi.ChainEpoch, types.TipSetKey) (*types.TipSet, error)
	ChainGetTipSet(context.Context, types.TipSetKey) (*types.TipSet, error)
	StateSearchMsg(ctx context.Context, from types.TipSetKey, msg cid.Cid, limit abi.ChainEpoch, allowReplaced bool) (*api.MsgLookup, error)
	ChainGetMessage(ctx context.Context, mc cid.Cid) (*types.Message, error)
}

type MessageWatcher 

type MessageWatcher struct {
	// contains filtered or unexported fields
}

func NewMessageWatcher 

func NewMessageWatcher(db *harmonydb.DB, ht *harmonytask.TaskEngine, pcs *chainsched.CurioChainSched, api MessageWaiterApi) (*MessageWatcher, error)

func (*MessageWatcher) AddOnLanded added in v1.28.3 

func (mw *MessageWatcher) AddOnLanded(fn func())

AddOnLanded registers a callback that fires after an update() pass landed at least one tracked message on chain. Use this to wake pollers (seal poller, deal market poller) that gate progress on after_*_msg_success or on the publish message reaching finality.

Callbacks are invoked asynchronously. Debouncing may be necessary.

Must be called before the first message lands; not safe to call concurrently with itself or with watcher updates.

func (*MessageWatcher) Stop 

func (mw *MessageWatcher) Stop(ctx context.Context) error

type MessageWatcherEth added in v1.25.1 

type MessageWatcherEth struct {
	// contains filtered or unexported fields
}

func NewMessageWatcherEth added in v1.25.1 

func NewMessageWatcherEth(db *harmonydb.DB, ht *harmonytask.TaskEngine, pcs *chainsched.CurioChainSched, api ethchain.EthClient) (*MessageWatcherEth, error)

func (*MessageWatcherEth) Stop added in v1.25.1 

func (mw *MessageWatcherEth) Stop(ctx context.Context) error

type PendingEthTx added in v1.28.3 

type PendingEthTx struct {
	WaitHash   string `db:"wait_hash"`
	LookupHash string `db:"lookup_hash"`
}

type Replacer added in v1.28.3 

type Replacer struct {
	// contains filtered or unexported fields
}

type ReplacerConfig added in v1.28.3 

type ReplacerConfig struct {
	DB         *harmonydb.DB
	ChainSched *chainsched.CurioChainSched

	Filecoin *FilecoinReplacerConfig
	Eth      *EthReplacerConfig
}

type SendTask 

type SendTask struct {
	// contains filtered or unexported fields
}

func (*SendTask) Adder 

func (s *SendTask) Adder(taskFunc harmonytask.AddTaskFunc)

func (*SendTask) CanAccept 

func (s *SendTask) CanAccept(ids []harmonytask.TaskID, engine *harmonytask.TaskEngine) ([]harmonytask.TaskID, error)

func (*SendTask) Do 

func (s *SendTask) Do(ctx context.Context, taskID harmonytask.TaskID, stillOwned func() bool) (done bool, err error)

func (*SendTask) TypeDetails 

func (s *SendTask) TypeDetails() harmonytask.TaskTypeDetails

type SendTaskETH added in v1.25.1 

type SendTaskETH struct {
	// contains filtered or unexported fields
}

func (*SendTaskETH) Adder added in v1.25.1 

func (s *SendTaskETH) Adder(taskFunc harmonytask.AddTaskFunc)

func (*SendTaskETH) CanAccept added in v1.25.1 

func (s *SendTaskETH) CanAccept(ids []harmonytask.TaskID, engine *harmonytask.TaskEngine) ([]harmonytask.TaskID, error)

func (*SendTaskETH) Do added in v1.25.1 

func (s *SendTaskETH) Do(ctx context.Context, taskID harmonytask.TaskID, stillOwned func() bool) (done bool, err error)

func (*SendTaskETH) TypeDetails added in v1.25.1 

func (s *SendTaskETH) TypeDetails() harmonytask.TaskTypeDetails

type Sender 

type Sender struct {
	// contains filtered or unexported fields
}

Sender abstracts away highly-available message sending with coordination through HarmonyDB. It make sure that nonces are assigned transactionally, and that messages are correctly broadcasted to the network. It ensures that messages are sent serially, and that failures to send don't cause nonce gaps.

func (*Sender) Send 

func (s *Sender) Send(ctx context.Context, msg *types.Message, mss *api.MessageSendSpec, reason string) (cid.Cid, error)

Send atomically assigns a nonce, signs, and pushes a message to mempool. maxFee is only used when GasFeeCap/GasPremium fields aren't specified

When maxFee is set to 0, Send will guess appropriate fee based on current chain conditions

Send behaves much like fullnodeApi.MpoolPushMessage, but it coordinates through HarmonyDB, making it safe to broadcast messages from multiple independent API nodes

Send is also currently more strict about required parameters than MpoolPushMessage

type SenderAPI 

type SenderAPI interface {
	StateAccountKey(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error)
	GasEstimateMessageGas(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec, tsk types.TipSetKey) (*types.Message, error)
	GasEstimateGasPremium(ctx context.Context, nblocksincl uint64, sender address.Address, gaslimit int64, tsk types.TipSetKey) (types.BigInt, error)
	WalletBalance(ctx context.Context, addr address.Address) (big.Int, error)
	MpoolGetNonce(context.Context, address.Address) (uint64, error)
	MpoolPush(context.Context, *types.SignedMessage) (cid.Cid, error)
}

type SenderETH added in v1.25.1 

type SenderETH struct {
	// contains filtered or unexported fields
}

func (*SenderETH) Send added in v1.25.1 

func (s *SenderETH) Send(ctx context.Context, fromAddress common.Address, tx *types.Transaction, reason string) (common.Hash, error)

Send sends an Ethereum transaction, coordinating nonce assignment, signing, and broadcasting.

func (*SenderETH) SendWithGasOverestimate added in v1.28.3 

func (s *SenderETH) SendWithGasOverestimate(ctx context.Context, fromAddress common.Address, tx *types.Transaction, reason string, overestimate float64) (common.Hash, error)

SendWithGasOverestimate is like Send, but multiplies the estimated gas limit by overestimate before submission (capped at the block gas limit).

type SignerAPI 

type SignerAPI interface {
	WalletSignMessage(context.Context, address.Address, *types.Message) (*types.SignedMessage, error)
}

type TaskEngine added in v1.26.0 

type TaskEngine interface {
	ResourcesAvailable() resources.Resources
}

TaskEngine is an interface for the parts of harmonytask.TaskEngine we need

Source Files

View all Source files

Directories

Path Synopsis
Package replace_mock is a generated GoMock package.
 Package replace_mock is a generated GoMock package.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.