structlog

Documentation

Index

• Constants
• func NewProcessBackwardsTask(payload *ProcessPayload) (*asynq.Task, error)
• func NewProcessForwardsTask(payload *ProcessPayload) (*asynq.Task, error)
• func NewVerifyBackwardsTask(payload *VerifyPayload) (*asynq.Task, error)
• func NewVerifyForwardsTask(payload *VerifyPayload) (*asynq.Task, error)
• type Config
  • func (c *Config) Validate() error
• type Dependencies
• type ProcessPayload
  • func (p *ProcessPayload) MarshalBinary() ([]byte, error)
  • func (p *ProcessPayload) UnmarshalBinary(data []byte) error
• type Processor
  • func New(ctx context.Context, deps *Dependencies, config *Config) (*Processor, error)
  • func (p *Processor) BatchInsertStructlogs(ctx context.Context, blockNumber uint64, transactionHash string, ...) error
  • func (p *Processor) EnqueueTask(ctx context.Context, task *asynq.Task, opts ...asynq.Option) error
  • func (p *Processor) GetHandlers() map[string]asynq.HandlerFunc
  • func (p *Processor) GetQueues() []c.QueueInfo
  • func (p *Processor) Name() string
  • func (p *Processor) ProcessNextBlock(ctx context.Context) error
  • func (p *Processor) ProcessSingleTransaction(ctx context.Context, block *types.Block, index int, tx *types.Transaction) (int, error)
  • func (p *Processor) SetProcessingMode(mode string)
  • func (p *Processor) Start(ctx context.Context) error
  • func (p *Processor) Stop(ctx context.Context) error
  • func (p *Processor) VerifyTransaction(ctx context.Context, blockNumber *big.Int, transactionHash string, ...) error
• type Structlog
• type VerifyPayload
  • func (v *VerifyPayload) MarshalBinary() ([]byte, error)
  • func (v *VerifyPayload) UnmarshalBinary(data []byte) error

Constants

const (
	ProcessorName            = "transaction_structlog"
	ProcessForwardsTaskType  = "transaction_structlog_process_forwards"
	ProcessBackwardsTaskType = "transaction_structlog_process_backwards"
	VerifyForwardsTaskType   = "transaction_structlog_verify_forwards"
	VerifyBackwardsTaskType  = "transaction_structlog_verify_backwards"
)

Functions

func NewProcessBackwardsTask

func NewProcessBackwardsTask(payload *ProcessPayload) (*asynq.Task, error)

NewProcessBackwardsTask creates a new backwards process task

func NewProcessForwardsTask

func NewProcessForwardsTask(payload *ProcessPayload) (*asynq.Task, error)

NewProcessForwardsTask creates a new forwards process task

func NewVerifyBackwardsTask

func NewVerifyBackwardsTask(payload *VerifyPayload) (*asynq.Task, error)

NewVerifyBackwardsTask creates a new backwards verify task

func NewVerifyForwardsTask

func NewVerifyForwardsTask(payload *VerifyPayload) (*asynq.Task, error)

NewVerifyForwardsTask creates a new forwards verify task

Types

type Config

type Config struct {
	clickhouse.Config `yaml:",inline"`
	Enabled           bool   `yaml:"enabled"`
	Table             string `yaml:"table"`
	BatchSize         int    `yaml:"batchSize" default:"10000"`
}

TransactionStructlogConfig holds configuration for transaction structlog processor

func (*Config) Validate

func (c *Config) Validate() error

type Dependencies

type Dependencies struct {
	Log         logrus.FieldLogger
	Pool        *ethereum.Pool
	Network     *ethereum.Network
	State       *state.Manager
	AsynqClient *asynq.Client
}

Dependencies contains the dependencies needed for the processor

type ProcessPayload

type ProcessPayload struct {
	BlockNumber      big.Int `json:"block_number"`
	TransactionHash  string  `json:"transaction_hash"`
	TransactionIndex uint32  `json:"transaction_index"`
	NetworkID        int32   `json:"network_id"`
	NetworkName      string  `json:"network_name"`
	Network          string  `json:"network"`         // Alias for NetworkName
	ProcessingMode   string  `json:"processing_mode"` // "forwards" or "backwards"
}

ProcessPayload represents the payload for processing a transaction

func (*ProcessPayload) MarshalBinary

func (p *ProcessPayload) MarshalBinary() ([]byte, error)

MarshalBinary implements encoding.BinaryMarshaler

func (*ProcessPayload) UnmarshalBinary

func (p *ProcessPayload) UnmarshalBinary(data []byte) error

UnmarshalBinary implements encoding.BinaryUnmarshaler

type Processor

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

Processor handles transaction structlog processing

func New

func New(ctx context.Context, deps *Dependencies, config *Config) (*Processor, error)

New creates a new transaction structlog processor

func (*Processor) BatchInsertStructlogs

func (p *Processor) BatchInsertStructlogs(ctx context.Context, blockNumber uint64, transactionHash string, transactionIndex uint32, structlogs []Structlog) error

BatchInsertStructlogs inserts structlog data in batch to ClickHouse

func (*Processor) EnqueueTask

func (p *Processor) EnqueueTask(ctx context.Context, task *asynq.Task, opts ...asynq.Option) error

EnqueueTask enqueues a task to the specified queue

func (*Processor) GetHandlers

func (p *Processor) GetHandlers() map[string]asynq.HandlerFunc

GetHandlers returns the task handlers for this processor

func (*Processor) GetQueues

func (p *Processor) GetQueues() []c.QueueInfo

GetQueues returns the queues used by this processor

func (*Processor) Name

func (p *Processor) Name() string

Name returns the processor name

func (*Processor) ProcessNextBlock

func (p *Processor) ProcessNextBlock(ctx context.Context) error

ProcessNextBlock processes the next available block

func (*Processor) ProcessSingleTransaction

func (p *Processor) ProcessSingleTransaction(ctx context.Context, block *types.Block, index int, tx *types.Transaction) (int, error)

ProcessSingleTransaction processes a single transaction (exposed for worker handlers)

func (*Processor) SetProcessingMode

func (p *Processor) SetProcessingMode(mode string)

SetProcessingMode sets the processing mode for the processor

func (*Processor) Start

func (p *Processor) Start(ctx context.Context) error

Start starts the processor

func (*Processor) Stop

func (p *Processor) Stop(ctx context.Context) error

Stop stops the processor

func (*Processor) VerifyTransaction

func (p *Processor) VerifyTransaction(ctx context.Context, blockNumber *big.Int, transactionHash string, transactionIndex uint32, networkName string, insertedCount int) error

VerifyTransaction verifies that a transaction has been processed correctly

type Structlog

type Structlog struct {
	UpdatedDateTime        time.Time `json:"updated_date_time"`
	BlockNumber            uint64    `json:"block_number"`
	TransactionHash        string    `json:"transaction_hash"`
	TransactionIndex       uint32    `json:"transaction_index"`
	TransactionGas         uint64    `json:"transaction_gas"`
	TransactionFailed      bool      `json:"transaction_failed"`
	TransactionReturnValue *string   `json:"transaction_return_value"`
	Index                  uint32    `json:"index"`
	ProgramCounter         uint32    `json:"program_counter"`
	Operation              string    `json:"operation"`
	Gas                    uint64    `json:"gas"`
	GasCost                uint64    `json:"gas_cost"`
	Depth                  uint64    `json:"depth"`
	ReturnData             *string   `json:"return_data"`
	Refund                 *uint64   `json:"refund"`
	Error                  *string   `json:"error"`
	MetaNetworkID          int32     `json:"meta_network_id"`
	MetaNetworkName        string    `json:"meta_network_name"`
}

type VerifyPayload

type VerifyPayload struct {
	BlockNumber      big.Int `json:"block_number"`
	TransactionHash  string  `json:"transaction_hash"`
	TransactionIndex uint32  `json:"transaction_index"`
	NetworkID        int32   `json:"network_id"`
	NetworkName      string  `json:"network_name"`
	Network          string  `json:"network"`         // Alias for NetworkName
	ProcessingMode   string  `json:"processing_mode"` // "forwards" or "backwards"
	InsertedCount    int     `json:"inserted_count"`
}

VerifyPayload represents the payload for verifying a transaction

func (*VerifyPayload) MarshalBinary

func (v *VerifyPayload) MarshalBinary() ([]byte, error)

MarshalBinary implements encoding.BinaryMarshaler

func (*VerifyPayload) UnmarshalBinary

func (v *VerifyPayload) UnmarshalBinary(data []byte) error

UnmarshalBinary implements encoding.BinaryUnmarshaler