Documentation
¶
Index ¶
- Variables
- type Config
- type DependencyGraphConfig
- type Service
- func (c *Service) BlockProcessing(stream servicepb.Coordinator_BlockProcessingServer) error
- func (c *Service) GetNextBlockNumberToCommit(ctx context.Context, _ *emptypb.Empty) (*servicepb.BlockRef, error)
- func (c *Service) GetTransactionsStatus(ctx context.Context, q *committerpb.TxIDsBatch) (*servicepb.TxStatusBatch, error)
- func (c *Service) NoPendingTransactionProcessing(context.Context, *emptypb.Empty) (*wrapperspb.BoolValue, error)
- func (c *Service) RegisterService(s serve.Servers)
- func (c *Service) Run(ctx context.Context) error
- func (c *Service) SetLastCommittedBlockNumber(ctx context.Context, lastBlock *servicepb.BlockRef) (*emptypb.Empty, error)
- func (c *Service) WaitForReady(ctx context.Context) bool
Constants ¶
This section is empty.
Variables ¶
var ( // ErrActiveStreamPendingTxProcessing is returned when NoPendingTransactionProcessing is called // while a stream is active. This value cannot be reliably determined in this state. ErrActiveStreamPendingTxProcessing = errors.New( "cannot check pending transaction processing while stream is active", ) // ErrExistingStreamOrConflictingOp indicates that a stream cannot be created because a stream already exists // or a conflicting gRPC API call is being made concurrently. ErrExistingStreamOrConflictingOp = errors.New("stream already exists or conflicting operation in progress") )
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
Verifier connection.MultiClientConfig `mapstructure:"verifier"`
ValidatorCommitter connection.MultiClientConfig `mapstructure:"validator-committer"`
DependencyGraph *DependencyGraphConfig `mapstructure:"dependency-graph" validate:"required"`
// ChannelBufferSizePerGoroutine defines the buffer size per go-routine.
ChannelBufferSizePerGoroutine int `mapstructure:"per-channel-buffer-size-per-goroutine" default:"10" validate:"gt=0"` //nolint:lll,revive
}
Config is the configuration for coordinator service. It contains configurations for all managers.
type DependencyGraphConfig ¶
type DependencyGraphConfig struct {
NumOfLocalDepConstructors int `mapstructure:"num-of-local-dep-constructors" default:"1" validate:"gt=0"`
WaitingTxsLimit int `mapstructure:"waiting-txs-limit" default:"100000" validate:"gt=0"`
// ChunkSize defines the maximum number of transactions to process in a single chunk for the dependency graph.
ChunkSize int `mapstructure:"chunk-size" default:"500" validate:"gt=0"`
}
DependencyGraphConfig is the configuration for dependency graph manager. It contains resource limits.
type Service ¶
type Service struct {
servicepb.UnimplementedCoordinatorServer
// contains filtered or unexported fields
}
Service is responsible for coordinating signature verification, dependency tracking, and validation and commit of each transaction.
func NewCoordinatorService ¶
NewCoordinatorService creates a new coordinator service.
func (*Service) BlockProcessing ¶
func (c *Service) BlockProcessing( stream servicepb.Coordinator_BlockProcessingServer, ) error
BlockProcessing receives a stream of blocks from the client and processes them.
func (*Service) GetNextBlockNumberToCommit ¶ added in v0.1.8
func (c *Service) GetNextBlockNumberToCommit( ctx context.Context, _ *emptypb.Empty, ) (*servicepb.BlockRef, error)
GetNextBlockNumberToCommit returns the next expected block number to be received by the coordinator.
func (*Service) GetTransactionsStatus ¶
func (c *Service) GetTransactionsStatus( ctx context.Context, q *committerpb.TxIDsBatch, ) (*servicepb.TxStatusBatch, error)
GetTransactionsStatus returns the status of given transactions identifiers.
func (*Service) NoPendingTransactionProcessing ¶ added in v1.0.2
func (c *Service) NoPendingTransactionProcessing( context.Context, *emptypb.Empty, ) (*wrapperspb.BoolValue, error)
NoPendingTransactionProcessing returns true when all previously submitted transactions have been processed (validated and committed by the VC) and the sidecar can safely reconnect. It is polled by the sidecar's waitForIdleCoordinator during recovery, while no stream is active.
Draining the status queue here is what breaks the deadlock described in issue #616. The status queue (vcServiceToCoordinatorTxStatus) is filled by the VC on the coordinator's lifetime context, but drained only by sendTxStatus, which lives and dies with the sidecar stream. While the stream is inactive, nothing drains the queue, yet the VC keeps forwarding statuses for the in-flight transactions. Once the queue fills, the VC writers block on it. If the number of in-flight transactions exceeds the queue capacity, readyCount plateaus below numTxsInProgress and can never catch up, so the equality below would never hold. The result is a circular wait: the sidecar cannot start a new stream (and thus resume draining) until this call reports idle, but this call cannot report idle until the VC — which is blocked on the full queue — finishes. By draining the queue on every call we free space for the blocked VC writers, let them finish producing, and let numTxsInProgress settle.
Discarding the drained statuses is safe: the VC produces a status only after the transaction is durably committed to the state DB, and the sidecar always recovers statuses from the DB (via GetTransactionsStatus), never from this queue. So while the stream is inactive these queued statuses have no consumer and can be dropped.
Accounting/invariants:
- numTxsInProgress is a superset of readyCount: every status counted in readyCount was previously counted in numTxsInProgress at block receipt, and numTxsInProgress is decremented only when a status leaves the queue (by sendTxStatus or by the drain below, both of which decrement readyCount by the same amount). Draining therefore preserves numTxsInProgress >= readyCount >= 0, so the result is never negative.
- Holding streamActive makes this mutually exclusive with the queue's only other reader, sendTxStatus (which runs under BlockProcessing's lock), and with receiveAndProcessBlock. VC writers may still run concurrently; that is safe because count is atomic and channels support concurrent send/receive.
- After draining, readyCount is ~0, so the equality reduces to numTxsInProgress == 0, i.e. every submitted transaction has produced (and we have consumed) its status. When that holds the queue is empty, so a reconnecting stream starts clean with no stale statuses.
func (*Service) RegisterService ¶ added in v0.1.5
RegisterService registers the coordinator's gRPC services and monitoring server.
Source Files
Directories