core

package
v0.0.0-...-1561025 Latest Latest
Warning

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

 Go to latest Published: Jan 25, 2026 License: Apache-2.0, MIT Imports: 6 Imported by: 0

Documentation

Overview

Package engine provides high-performance transaction processing. This package implements: - Worker pool with goroutines (replacing Python's ThreadPool) - Transaction ordering service - Mempool management

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrMempoolFull     = errors.New("mempool is full")
	ErrTxAlreadyExists = errors.New("transaction already exists")
	ErrTxNotFound      = errors.New("transaction not found")
	ErrInvalidTx       = errors.New("invalid transaction")
)

Common errors for mempool operations

Functions

This section is empty.

Types

type BlockBuilder 

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

BlockBuilder batches certified events into blocks.

func NewBlockBuilder 

func NewBlockBuilder(blockSize int, timeout time.Duration) *BlockBuilder

NewBlockBuilder creates a new block builder.

func (*BlockBuilder) AddEvent 

func (b *BlockBuilder) AddEvent(event *PendingEvent) []*PendingEvent

AddEvent adds a certified event to the current batch. Returns the batch if ready for block creation, nil otherwise.

func (*BlockBuilder) BatchSize 

func (b *BlockBuilder) BatchSize() int

BatchSize returns current batch size.

func (*BlockBuilder) ForceFlush 

func (b *BlockBuilder) ForceFlush() []*PendingEvent

ForceFlush forces block creation from current batch.

type Certification 

type Certification struct {
	EventID  string
	Valid    bool
	Errors   []string
	CertAt   time.Time
	Metadata map[string]interface{}
}

Certification contains validation result for an event.

type EventCertifier 

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

EventCertifier validates events before ordering.

func NewEventCertifier 

func NewEventCertifier() *EventCertifier

NewEventCertifier creates a new event certifier.

func (*EventCertifier) AddRule 

func (c *EventCertifier) AddRule(rule ValidationRule)

AddRule registers a validation rule.

func (*EventCertifier) GetCertification 

func (c *EventCertifier) GetCertification(eventID string) *Certification

GetCertification retrieves a certification by event ID.

func (*EventCertifier) Validate 

func (c *EventCertifier) Validate(event *PendingEvent) *Certification

Validate validates an event and returns certification result.

type EventStatus 

type EventStatus int

EventStatus represents the processing status of an event. 

const (
	EventPending EventStatus = iota
	EventProcessing
	EventOrdered
	EventCertified
	EventRejected
)

func (EventStatus) String 

func (s EventStatus) String() string

type Mempool 

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

Mempool manages pending transactions with thread-safe operations.

func NewMempool 

func NewMempool(maxSize int) *Mempool

NewMempool creates a new Mempool with the specified maximum size.

func (*Mempool) Add 

func (m *Mempool) Add(tx *Transaction) error

Add adds a transaction to the mempool. Returns error if mempool is full or transaction already exists.

func (*Mempool) Clear 

func (m *Mempool) Clear()

Clear removes all transactions from the mempool.

func (*Mempool) Contains 

func (m *Mempool) Contains(txID string) bool

Contains checks if a transaction exists in the mempool.

func (*Mempool) Get 

func (m *Mempool) Get(txID string) *Transaction

Get retrieves a transaction by ID without removing it.

func (*Mempool) IsFull 

func (m *Mempool) IsFull() bool

IsFull returns true if the mempool has reached its maximum size.

func (*Mempool) Peek 

func (m *Mempool) Peek(n int) []*Transaction

Peek returns up to n highest-priority transactions without removing them.

func (*Mempool) PopBatch 

func (m *Mempool) PopBatch(n int) []*Transaction

PopBatch removes and returns up to n highest-priority transactions.

func (*Mempool) Remove 

func (m *Mempool) Remove(txID string) bool

Remove removes a transaction by ID. Returns true if the transaction was found and removed.

func (*Mempool) Size 

func (m *Mempool) Size() int

Size returns the current number of transactions in the mempool.

func (*Mempool) Stats 

func (m *Mempool) Stats() MempoolStats

type MempoolStats 

type MempoolStats struct {
	Size      int `json:"size"`
	MaxSize   int `json:"max_size"`
	Available int `json:"available"`
}

Stats returns mempool statistics.

type OrderingConfig 

type OrderingConfig struct {
	BlockSize    int
	BatchTimeout time.Duration
	Workers      int
	MaxPending   int
}

OrderingConfig contains configuration for the ordering service.

func DefaultOrderingConfig 

func DefaultOrderingConfig() OrderingConfig

DefaultOrderingConfig returns default configuration.

type OrderingService 

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

OrderingService coordinates event ordering and block creation.

func NewOrderingService 

func NewOrderingService(config OrderingConfig) *OrderingService

NewOrderingService creates a new ordering service.

func (*OrderingService) Blocks 

func (s *OrderingService) Blocks() <-chan []*PendingEvent

Blocks returns the channel for receiving completed blocks.

func (*OrderingService) GetStats 

func (s *OrderingService) GetStats() OrderingStats

GetStats returns service statistics.

func (*OrderingService) GetStatus 

func (s *OrderingService) GetStatus() OrderingStatus

GetStatus returns current service status.

func (*OrderingService) Start 

func (s *OrderingService) Start() error

Start begins the ordering service.

func (*OrderingService) Stop 

func (s *OrderingService) Stop()

Stop stops the ordering service.

func (*OrderingService) SubmitEvent 

func (s *OrderingService) SubmitEvent(event *PendingEvent) error

SubmitEvent submits an event for ordering.

type OrderingStats 

type OrderingStats struct {
	Status          string `json:"status"`
	EventsReceived  int64  `json:"events_received"`
	EventsCertified int64  `json:"events_certified"`
	EventsRejected  int64  `json:"events_rejected"`
	BlocksCreated   int64  `json:"blocks_created"`
	PendingCount    int    `json:"pending_count"`
	BatchSize       int    `json:"current_batch_size"`
}

OrderingStats contains service statistics.

type OrderingStatus 

type OrderingStatus int

OrderingStatus represents the status of the ordering service. 

const (
	StatusActive OrderingStatus = iota
	StatusMaintenance
	StatusLockdown
	StatusShutdown
	StatusError
)

func (OrderingStatus) String 

func (s OrderingStatus) String() string

type PendingEvent 

type PendingEvent struct {
	ID         string
	Data       map[string]interface{}
	ChannelID  string
	Submitter  string
	ReceivedAt time.Time
	Status     EventStatus
	Cert       *Certification
}

PendingEvent represents an event waiting to be ordered.

type PoolStats 

type PoolStats struct {
	Name        string  `json:"name"`
	Workers     int     `json:"workers"`
	Active      int64   `json:"active"`
	Completed   int64   `json:"completed"`
	Failed      int64   `json:"failed"`
	Pending     int     `json:"pending"`
	SuccessRate float64 `json:"success_rate"`
}

PoolStats contains worker pool statistics.

type Result 

type Result struct {
	TaskID   string
	Success  bool
	Data     interface{}
	Error    error
	Duration time.Duration
	WorkerID int
}

Result represents the result of task processing.

type Task 

type Task struct {
	ID          string
	Data        interface{}
	ProcessFunc func(interface{}) (interface{}, error)
	Priority    int
	CreatedAt   time.Time
	Ctx         context.Context
}

Task represents a processing task for the worker pool.

func NewTask 

func NewTask(id string, data interface{}, fn func(interface{}) (interface{}, error)) *Task

NewTask creates a new task with default values.

type Transaction 

type Transaction struct {
	ID        string                 `json:"id"`
	EntityID  string                 `json:"entity_id"`
	EventType string                 `json:"event_type"`
	Data      []byte                 `json:"data,omitempty"`
	Priority  int                    `json:"priority"`
	Timestamp time.Time              `json:"timestamp"`
	Metadata  map[string]interface{} `json:"metadata,omitempty"`
}

Transaction represents a pending transaction in the mempool.

func (*Transaction) Validate 

func (tx *Transaction) Validate() error

Validate checks if the transaction has required fields.

type ValidationRule 

type ValidationRule func(data map[string]interface{}) error

ValidationRule is a function that validates event data.

type WorkerPool 

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

WorkerPool manages a pool of goroutine workers for parallel processing.

func NewWorkerPool 

func NewWorkerPool(name string, workers int) *WorkerPool

NewWorkerPool creates a new worker pool with the specified number of workers.

func (*WorkerPool) GetStats 

func (p *WorkerPool) GetStats() PoolStats

GetStats returns current worker pool statistics.

func (*WorkerPool) IsRunning 

func (p *WorkerPool) IsRunning() bool

IsRunning returns true if the pool is still accepting tasks.

func (*WorkerPool) Results 

func (p *WorkerPool) Results() <-chan *Result

Results returns the result channel for consuming results.

func (*WorkerPool) Shutdown 

func (p *WorkerPool) Shutdown()

Shutdown gracefully shuts down the worker pool.

func (*WorkerPool) ShutdownWithTimeout 

func (p *WorkerPool) ShutdownWithTimeout(timeout time.Duration) error

ShutdownWithTimeout shuts down with a timeout.

func (*WorkerPool) Submit 

func (p *WorkerPool) Submit(task *Task) error

Submit adds a task to the worker pool for processing.

func (*WorkerPool) SubmitAndWait 

func (p *WorkerPool) SubmitAndWait(task *Task, timeout time.Duration) (*Result, error)

SubmitAndWait submits a task and waits for its result.

Source Files

View all Source files

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.