protocol

package
v0.0.0-...-964a6ab Latest Latest
Warning

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

 Go to latest Published: Oct 9, 2026 License: MIT Imports: 38 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	BaseError int = iota
	NoAvailableUpstreams
	WrongChain
	CtxErrorCode
	SubscribeTotalFailure
	NoApiConnectors
	ClientErrorCode         = 400
	AuthErrorCode           = 403
	RequestTimeout          = 408
	InternalServerErrorCode = 500
	RateLimitExceeded       = 429
	NoSupportedMethod       = -32601
	IncorrectResponseBody   = -32001
	QuorumSignatureErrCode  = -32010
)
View Source 
const GrpcErrorCodeBase = 10000

GrpcErrorCodeBase offsets canonical gRPC codes (1-16) out of the protocol's own error-code space (the iota block 0-5, the HTTP-ish 4xx/5xx and the JSON-RPC -32xxx constants), so no ResponseError.Code consumer can ever alias a gRPC code with a nodecore code. The int is informational (logs, encoded error bodies); everything that routes or renders gRPC errors reads the typed GrpcStatus in Data instead - the client-facing status is reconstructed from it, never from this int.

View Source 
const InvalidParams = -32602
View Source 
const MaxChunkSize = 8192
View Source 
const MethodSeparator = "#"
View Source 
const NoResultStart = -1

NoResultStart is the ResultStart value meaning no "result" value was located in the first chunk. Consumers must treat it as "do not unwrap" rather than as an offset - the zero value would silently mean offset 0.

Variables

View Source 
var NoJsonRpcResultStreamHint = JsonRpcResultStreamHint{ResultStart: NoResultStart}

NoJsonRpcResultStreamHint is the hint of a response that carries no unwrapping information; unwrapping it fails instead of truncating.

Functions

func GrpcStatusOf 

func GrpcStatusOf(respError *ResponseError) (grpcStatus *status.Status, fromStatusProto bool)

GrpcStatusOf turns a flow error into the *status.Status a gRPC client receives. An upstream gRPC status rides through verbatim - typed details included; nodecore's own error codes are mapped onto the closed 17-code model. Shared by every gRPC-facing ingress so no client needs its own table.

fromStatusProto reports that the status was reconstructed from the carried google.rpc.Status bytes, i.e. those bytes are valid and may go on the wire verbatim; false covers nodecore errors, statuses without proto bytes, and bytes that failed to unmarshal (the code/message fallback).

func IsGrpcErrorNotRetryable 

func IsGrpcErrorNotRetryable(respError *ResponseError) bool

IsGrpcErrorNotRetryable reports whether an error is a gRPC status whose retryability was already decided at construction time by the closed code model (retryable codes become partial failures; everything else must NOT be re-litigated by the JSON-RPC message matching in errors_config - upstream status messages like "transaction not found" would falsely match).

func IsGrpcRateLimited 

func IsGrpcRateLimited(response ResponseHolder) bool

IsGrpcRateLimited reports whether a response is an upstream RESOURCE_EXHAUSTED, the gRPC analog of HTTP 429 for rate-limit accounting.

func IsRetryable 

func IsRetryable(response ResponseHolder) bool

func IsStream 

func IsStream(method string) bool

func LabelCacheKey 

func LabelCacheKey(selectors []RequestSelector) string

LabelCacheKey reduces a selector tree to its node-class identity for cache keying. Some chains expose distinct node classes (selected via label selectors) that return different responses for the same method+params, so the chosen class must be part of the cache key or one class's response can be served to a request that pinned another.

Only label/exists selectors carry a node-class constraint; every other selector (height, block tag, slot, lower bound, any, unsupported) is treated as ⊤ — it admits every class and is irrelevant to which class answers. The reduction walks the boolean structure over that label dimension only: AND intersects (⊤ children drop out), OR unions (any ⊤ child makes the whole union ⊤), NOT complements. An empty result means the request is unconstrained over node class and shares the default cache entry, leaving keys for label-free traffic byte-identical to before.

func LagExceeds 

func LagExceeds(lag, syncingLag int64) bool

LagExceeds reports whether a head lag is large enough to matter. A non-positive threshold means "not configured" and disables the check, so an upstream is never downgraded on a chain that has no syncing threshold.

func ResponseMetadata 

func ResponseMetadata(response any) (http.Header, map[string][]string)

ResponseMetadata reads the optional transport metadata of a response through the two capabilities above. It takes any because callers hold values behind different interfaces (ResponseHolder, SubResponse).

func ResultAsNumber 

func ResultAsNumber(result []byte) uint64

func ResultAsString 

func ResultAsString(result []byte) string

func ToHttpCode 

func ToHttpCode(response ResponseHolder) int

Types

type AbstractUpstreamStateEvent 

type AbstractUpstreamStateEvent interface {
	ProcessEvent(state UpstreamState) UpstreamState
	Same(state UpstreamState) bool
}

type AvailabilityStatus 

type AvailabilityStatus int
const (
	Available AvailabilityStatus = iota + 1
	Immature
	Syncing
	Unavailable

	UnknownStatus = math.MaxInt
)

func StatusByLag 

func StatusByLag(lag int64, avail AvailabilityStatus, syncingLag int64) AvailabilityStatus

func (AvailabilityStatus) String 

func (a AvailabilityStatus) String() string

type BanMethodUpstreamStateEvent 

type BanMethodUpstreamStateEvent struct {
	Method string
}

func (*BanMethodUpstreamStateEvent) ProcessEvent 

func (b *BanMethodUpstreamStateEvent) ProcessEvent(state UpstreamState) UpstreamState

func (*BanMethodUpstreamStateEvent) Same 

func (b *BanMethodUpstreamStateEvent) Same(_ UpstreamState) bool

type Block 

type Block struct {
	Height     uint64
	Slot       uint64
	Hash       blockchain.HashId
	ParentHash blockchain.HashId
	RawData    []byte
}

func NewBlock 

func NewBlock(height, slot uint64, hash, parentHash blockchain.HashId) Block

func NewBlockWithHeight 

func NewBlockWithHeight(height uint64) Block

func NewBlockWithHeights 

func NewBlockWithHeights(height, slot uint64) Block

func (Block) CompareWithHeight 

func (b Block) CompareWithHeight(other Block) int

func (Block) Equals 

func (b Block) Equals(other Block) bool

func (Block) IsEmptyByHeight 

func (b Block) IsEmptyByHeight() bool

func (Block) IsFullEmpty 

func (b Block) IsFullEmpty() bool

type BlockInfo 

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

func NewBlockInfo 

func NewBlockInfo() *BlockInfo

func (*BlockInfo) AddBlock 

func (b *BlockInfo) AddBlock(data Block, blockType BlockType)

func (*BlockInfo) Copy 

func (b *BlockInfo) Copy() *BlockInfo

func (*BlockInfo) GetBlock 

func (b *BlockInfo) GetBlock(blockType BlockType) Block

func (*BlockInfo) GetBlocks 

func (b *BlockInfo) GetBlocks() map[BlockType]Block

type BlockType 

type BlockType int
const (
	FinalizedBlock BlockType = iota
	SafeBlock
)

func (BlockType) String 

func (b BlockType) String() string

type BlockUpstreamStateEvent 

type BlockUpstreamStateEvent struct {
	Block     Block
	BlockType BlockType
}

func (*BlockUpstreamStateEvent) ProcessEvent 

func (b *BlockUpstreamStateEvent) ProcessEvent(state UpstreamState) UpstreamState

func (*BlockUpstreamStateEvent) Same 

func (b *BlockUpstreamStateEvent) Same(state UpstreamState) bool

type Cap 

type Cap int
const (
	// WsCap means the upstream has a live websocket connector (any chain). Used
	// to route node-backed subscriptions to a ws-capable upstream.
	WsCap Cap = iota
	// NewHeadsCap means the upstream's head is subscription-driven (its head
	// connector is a websocket), so newHeads can be synthesized locally from the
	// head stream. EVM-only.
	NewHeadsCap
	// LogsCap means NewHeadsCap holds and eth_getLogs is enabled, so logs can be
	// synthesized locally. EVM-only.
	LogsCap
	// PendingTxCap means the upstream has a live websocket connector usable for
	// eth_subscribe("newPendingTransactions"), so pending-tx subscriptions can be
	// aggregated locally. Governed by the cap pipeline: a live ws connector is enough
	// on most chains, but on chains with a pending-tx validator (e.g. BASE) it is also
	// gated on txpool_content showing a non-trivial mempool. The topics are EVM-only
	// but the cap is harmless elsewhere since only eth_subscribe clients request them.
	PendingTxCap
)

type CapsUpstreamStateEvent 

type CapsUpstreamStateEvent struct {
	Caps mapset.Set[Cap]
}

CapsUpstreamStateEvent replaces the upstream's capability set wholesale. Caps are produced by the per-chain cap pipeline (caps.CapProcessor aggregating CapDetectors), which already merges every detector's view into one set, so the event carries the full set rather than incremental add/remove.

func (*CapsUpstreamStateEvent) ProcessEvent 

func (c *CapsUpstreamStateEvent) ProcessEvent(state UpstreamState) UpstreamState

func (*CapsUpstreamStateEvent) Same 

func (c *CapsUpstreamStateEvent) Same(state UpstreamState) bool

type ClientRetryableError 

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

func NewClientRetryableError 

func NewClientRetryableError(err error) *ClientRetryableError

func (ClientRetryableError) Error 

func (e ClientRetryableError) Error() string

type CloseReader 

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

func NewCloseReader 

func NewCloseReader(ctx context.Context, mainReader io.Reader, readerToClose io.ReadCloser) *CloseReader

func (*CloseReader) Close 

func (c *CloseReader) Close() error

Close closes the underlying body. It is safe to call multiple times and concurrently with Read: a streaming consumer that aborts early (e.g. the gRPC client disconnected) can call Close to unblock a read-ahead goroutine parked in Read on a slow upstream, so it returns deterministically instead of lingering until the HTTP client timeout fires.

func (*CloseReader) Read 

func (c *CloseReader) Read(p []byte) (n int, err error)

type FatalErrorUpstreamStateEvent 

type FatalErrorUpstreamStateEvent struct{}

func (*FatalErrorUpstreamStateEvent) ProcessEvent 

func (f *FatalErrorUpstreamStateEvent) ProcessEvent(state UpstreamState) UpstreamState

func (*FatalErrorUpstreamStateEvent) Same 

func (f *FatalErrorUpstreamStateEvent) Same(_ UpstreamState) bool

type FirstChunkAnalysis 

type FirstChunkAnalysis struct {
	// Streamable is true iff a "result" value was located AND no
	// envelope-level "error" key was seen in the first chunk.
	Streamable bool
	// ResultStart is the byte offset of the "result" value within the first
	// chunk, or NoResultStart if none was found.
	ResultStart int
	// Counter is primed for the "result" value's JSON type (valid only when
	// Streamable).
	Counter ResultCounter
}

FirstChunkAnalysis is the result of inspecting the first chunk of a JSON-RPC response in a single pass. It both decides whether the response may be streamed and, for the gRPC result-unwrap path, hands the consumer the offset and primed counter for the "result" value so it does not have to re-scan the chunk.

func AnalyzeChunk 

func AnalyzeChunk(buf []byte) FirstChunkAnalysis

AnalyzeChunk performs the single-pass inspection of an already-materialized first chunk. AnalyzeFirstChunk wraps it with a bufio Peek; it is exported so the gRPC consumer's tests can locate a result value in a chunk without a reader. See AnalyzeFirstChunk for the walk's semantics.

func AnalyzeFirstChunk 

func AnalyzeFirstChunk(reader *bufio.Reader, chunkSize int) FirstChunkAnalysis

AnalyzeFirstChunk peeks the first chunk of a JSON-RPC response and inspects it in a single json.Decoder walk (mirroring dshackle's parseFirstPart):

  • an envelope-level "error" key means the response must NOT be streamed (the buffered path parses it so the flow can fail over);
  • the "result" value's start offset and JSON type are recorded so the gRPC consumer can emit it without a second scan;
  • after "result" is located its value is skipped and the rest of the chunk is still scanned so a trailing "error" key is caught too.

The walk tolerates a truncated chunk (the result value routinely overflows MaxChunkSize): a token error simply ends the walk.

func (FirstChunkAnalysis) Hint 

func (a FirstChunkAnalysis) Hint() StreamHint

Hint returns a StreamHint to attach to a streaming response, or nil when the analysis found nothing streamable (so the consumer must not try to unwrap).

type GenericSubResponse 

type GenericSubResponse struct {
	Message     []byte
	Error       *ResponseError
	UpstreamId  string
	ParsedEvent ParsedEvent
}

GenericSubResponse is the transport-neutral subscription event synthesized by the generic pipeline itself (sources, subengine, heads). Transport layers push their own SubResponse implementations instead.

func (*GenericSubResponse) GetError 

func (g *GenericSubResponse) GetError() *ResponseError

func (*GenericSubResponse) GetMessage 

func (g *GenericSubResponse) GetMessage() []byte

func (*GenericSubResponse) GetParsedEvent 

func (g *GenericSubResponse) GetParsedEvent() ParsedEvent

func (*GenericSubResponse) GetUpstreamId 

func (g *GenericSubResponse) GetUpstreamId() string

func (*GenericSubResponse) IsEnd 

func (g *GenericSubResponse) IsEnd() bool

IsEnd is false: synthesized sources end by closing their channel.

type GenericUpstreamResponse 

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

func NewGrpcUpstreamResponse 

func NewGrpcUpstreamResponse(id string, body []byte) *GenericUpstreamResponse

NewGrpcUpstreamResponse frames a successful unary gRPC reply: the response message bytes verbatim, never parsed.

func NewHttpUpstreamResponse 

func NewHttpUpstreamResponse(id string, body []byte, responseCode int, requestType RequestType) *GenericUpstreamResponse

func NewHttpUpstreamResponseStream 

func NewHttpUpstreamResponseStream(id string, reader io.Reader, requestType RequestType) *GenericUpstreamResponse

func NewHttpUpstreamResponseWithError 

func NewHttpUpstreamResponseWithError(error *ResponseError) *GenericUpstreamResponse

func NewSimpleHttpUpstreamResponse 

func NewSimpleHttpUpstreamResponse(id string, body []byte, requestType RequestType) *GenericUpstreamResponse

func (*GenericUpstreamResponse) EncodeResponse 

func (h *GenericUpstreamResponse) EncodeResponse(realId []byte) io.Reader

func (*GenericUpstreamResponse) GetError 

func (h *GenericUpstreamResponse) GetError() *ResponseError

func (*GenericUpstreamResponse) GetStreamHint 

func (h *GenericUpstreamResponse) GetStreamHint() StreamHint

GetStreamHint returns the streaming hint, or nil if none was recorded.

func (*GenericUpstreamResponse) HasError 

func (h *GenericUpstreamResponse) HasError() bool

func (*GenericUpstreamResponse) HasStream 

func (h *GenericUpstreamResponse) HasStream() bool

func (*GenericUpstreamResponse) Id 

func (h *GenericUpstreamResponse) Id() string

func (*GenericUpstreamResponse) ResponseCode 

func (h *GenericUpstreamResponse) ResponseCode() int

func (*GenericUpstreamResponse) ResponseHeaders 

func (h *GenericUpstreamResponse) ResponseHeaders() http.Header

func (*GenericUpstreamResponse) ResponseResult 

func (h *GenericUpstreamResponse) ResponseResult() []byte

func (*GenericUpstreamResponse) ResponseResultString 

func (h *GenericUpstreamResponse) ResponseResultString() (string, error)

func (*GenericUpstreamResponse) ResponseTrailers 

func (h *GenericUpstreamResponse) ResponseTrailers() map[string][]string

func (*GenericUpstreamResponse) WithResponseHeaders 

func (h *GenericUpstreamResponse) WithResponseHeaders(headers http.Header) *GenericUpstreamResponse

func (*GenericUpstreamResponse) WithResponseTrailers 

func (h *GenericUpstreamResponse) WithResponseTrailers(trailers map[string][]string) *GenericUpstreamResponse

func (*GenericUpstreamResponse) WithStreamHint 

func (h *GenericUpstreamResponse) WithStreamHint(hint StreamHint) *GenericUpstreamResponse

WithStreamHint attaches the first-chunk analysis produced by AnalyzeFirstChunk so the gRPC result-unwrap consumer can emit the result without re-scanning the chunk. A nil hint leaves the response without one.

type GrpcStatus 

type GrpcStatus struct {
	Code    codes.Code
	Message string
	// StatusProto is the serialized google.rpc.Status when the upstream
	// attached typed details; nil otherwise. Kept serialized so the protocol
	// layer never parses it.
	StatusProto []byte
}

GrpcStatus is the upstream's verbatim gRPC status. It rides through the flow as ResponseError.Data so the ingress can hand the client exactly what the upstream said, while the classification below drives routing independently (the two-track rule).

func GrpcStatusFromError 

func GrpcStatusFromError(respError *ResponseError) (*GrpcStatus, bool)

GrpcStatusFromError extracts the verbatim upstream status a gRPC error response carries, if any.

type GrpcSubResponse 

type GrpcSubResponse struct {
	Message    []byte
	Error      *ResponseError
	End        bool
	UpstreamId string
	Headers    http.Header
	Trailers   map[string][]string
}

GrpcSubResponse is one frame of an upstream gRPC server stream: a message (Message), the terminal status (Error, carrying the verbatim GrpcStatus), or the clean end of the stream (End). Headers ride on the first frame, Trailers on the terminal one (error or end frame).

func (*GrpcSubResponse) GetError 

func (g *GrpcSubResponse) GetError() *ResponseError

func (*GrpcSubResponse) GetMessage 

func (g *GrpcSubResponse) GetMessage() []byte

func (*GrpcSubResponse) GetParsedEvent 

func (g *GrpcSubResponse) GetParsedEvent() ParsedEvent

GetParsedEvent is nil: gRPC frames are opaque bytes, never parsed.

func (*GrpcSubResponse) GetUpstreamId 

func (g *GrpcSubResponse) GetUpstreamId() string

func (*GrpcSubResponse) IsEnd 

func (g *GrpcSubResponse) IsEnd() bool

func (*GrpcSubResponse) ResponseHeaders 

func (g *GrpcSubResponse) ResponseHeaders() http.Header

func (*GrpcSubResponse) ResponseTrailers 

func (g *GrpcSubResponse) ResponseTrailers() map[string][]string

type GrpcUpstreamSubscriptionResponse 

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

GrpcUpstreamSubscriptionResponse is the handle GrpcConnector.Subscribe returns: the frame channel plus an op id (informational - the stream's lifetime is the ctx passed to Subscribe, so Unsubscribe is a no-op for gRPC).

func NewGrpcUpstreamSubscriptionResponse 

func NewGrpcUpstreamSubscriptionResponse(messages chan SubResponse, opId string) *GrpcUpstreamSubscriptionResponse

func (*GrpcUpstreamSubscriptionResponse) OpId 

func (g *GrpcUpstreamSubscriptionResponse) OpId() string

func (*GrpcUpstreamSubscriptionResponse) ResponseChan 

func (g *GrpcUpstreamSubscriptionResponse) ResponseChan() chan SubResponse

type HasResponseHeaders 

type HasResponseHeaders interface {
	ResponseHeaders() http.Header
}

HasResponseHeaders is an optional capability for response holders that carry upstream HTTP response headers (e.g. QR<N>-id-* quorum signatures). Kept separate from ResponseHolder to avoid forcing every implementation.

type HasResponseTrailers 

type HasResponseTrailers interface {
	ResponseTrailers() map[string][]string
}

HasResponseTrailers is an optional capability for response holders that carry upstream gRPC trailer metadata. Only the gRPC ingress consumes it - a gRPC client must receive trailers *as* trailers, not folded into the initial metadata. Keys follow gRPC metadata convention (lowercase).

type HeadUpstreamEvent 

type HeadUpstreamEvent struct {
	Status AvailabilityStatus
	Head   Block
}

type HeadUpstreamStateEvent 

type HeadUpstreamStateEvent struct {
	HeadData Block
}

func (*HeadUpstreamStateEvent) ProcessEvent 

func (h *HeadUpstreamStateEvent) ProcessEvent(state UpstreamState) UpstreamState

func (*HeadUpstreamStateEvent) Same 

func (h *HeadUpstreamStateEvent) Same(_ UpstreamState) bool

type HttpMethod 

type HttpMethod int
const (
	Get HttpMethod = iota
	Post
)

func (HttpMethod) String 

func (h HttpMethod) String() string

type InitUpstreamStateEvent 

type InitUpstreamStateEvent struct {
}

func (*InitUpstreamStateEvent) ProcessEvent 

func (i *InitUpstreamStateEvent) ProcessEvent(state UpstreamState) UpstreamState

func (*InitUpstreamStateEvent) Same 

func (i *InitUpstreamStateEvent) Same(state UpstreamState) bool

type JsonRpcRequestBody 

type JsonRpcRequestBody struct {
	Id      json.RawMessage `json:"id"`
	Jsonrpc string          `json:"jsonrpc"`
	Method  string          `json:"method"`
	Params  json.RawMessage `json:"params"`
}

type JsonRpcResultStreamHint 

type JsonRpcResultStreamHint struct {
	ResultStart int
	Counter     ResultCounter
}

JsonRpcResultStreamHint is the StreamHint for unwrapping the "result" value of a JSON-RPC envelope: where the value begins in the first chunk and a counter primed for its JSON type so the consumer can detect where it ends.

func (JsonRpcResultStreamHint) Type 

func (JsonRpcResultStreamHint) Type() RequestType

type JsonRpcWsUpstreamResponse 

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

func NewJsonRpcWsUpstreamResponse 

func NewJsonRpcWsUpstreamResponse(messages chan SubResponse, subOpId string) *JsonRpcWsUpstreamResponse

func (*JsonRpcWsUpstreamResponse) OpId 

func (j *JsonRpcWsUpstreamResponse) OpId() string

func (*JsonRpcWsUpstreamResponse) ResponseChan 

func (j *JsonRpcWsUpstreamResponse) ResponseChan() chan SubResponse

type Labels 

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

func NewLabels 

func NewLabels() *Labels

func (*Labels) AddLabel 

func (l *Labels) AddLabel(key, value string)

func (*Labels) Copy 

func (l *Labels) Copy() *Labels

func (*Labels) GetAllLabels 

func (l *Labels) GetAllLabels() map[string]string

func (*Labels) GetLabel 

func (l *Labels) GetLabel(label string) (string, bool)

type LabelsUpstreamStateEvent 

type LabelsUpstreamStateEvent struct {
	Labels lo.Tuple2[string, string]
}

func (*LabelsUpstreamStateEvent) ProcessEvent 

func (l *LabelsUpstreamStateEvent) ProcessEvent(state UpstreamState) UpstreamState

func (*LabelsUpstreamStateEvent) Same 

func (l *LabelsUpstreamStateEvent) Same(state UpstreamState) bool

type LowerBoundData 

type LowerBoundData struct {
	Bound     int64
	Timestamp int64
	Type      LowerBoundType
}

func NewLowerBoundData 

func NewLowerBoundData(bound, timestamp int64, boundType LowerBoundType) LowerBoundData

func NewLowerBoundDataNow 

func NewLowerBoundDataNow(bound int64, boundType LowerBoundType) LowerBoundData

type LowerBoundInfo 

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

func NewLowerBoundInfo 

func NewLowerBoundInfo() *LowerBoundInfo

func (*LowerBoundInfo) AddLowerBound 

func (l *LowerBoundInfo) AddLowerBound(data LowerBoundData)

func (*LowerBoundInfo) Copy 

func (l *LowerBoundInfo) Copy() *LowerBoundInfo

func (*LowerBoundInfo) GetAllBounds 

func (l *LowerBoundInfo) GetAllBounds() []LowerBoundData

func (*LowerBoundInfo) GetLowerBound 

func (l *LowerBoundInfo) GetLowerBound(boundType LowerBoundType) (LowerBoundData, bool)

type LowerBoundType 

type LowerBoundType int
const (
	UnknownBound LowerBoundType = iota + 1
	SlotBound
	StateBound
	ReceiptsBound
	TxBound
	BlockBound
	LogsBound
	TraceBound
	ProofBound
	EpochBound
	BlobBound
)

func ParseLowerBoundType 

func ParseLowerBoundType(name string) (LowerBoundType, bool)

ParseLowerBoundType maps a case-insensitive type name to its LowerBoundType. It never returns UnknownBound.

func (LowerBoundType) String 

func (t LowerBoundType) String() string

type LowerBoundUpstreamStateEvent 

type LowerBoundUpstreamStateEvent struct {
	Data LowerBoundData
}

func (*LowerBoundUpstreamStateEvent) ProcessEvent 

func (l *LowerBoundUpstreamStateEvent) ProcessEvent(state UpstreamState) UpstreamState

func (*LowerBoundUpstreamStateEvent) Same 

func (l *LowerBoundUpstreamStateEvent) Same(state UpstreamState) bool

Same also filters out updates that must not be applied (the event loop skips ProcessEvent when Same is true): a bound above the upstream head, or a value lower than the current one unless it is the archive reset (1). These are the LowerBoundProcessor.processBounds rules duplicated for live corrections (request_processor.liveLowerBoundFromPrunedError), which bypass the processor; otherwise the detector's next publish would undo them. TODO: route live corrections through LowerBoundProcessor and drop this duplication.

type MethodAvailability 

type MethodAvailability int

MethodAvailability is what an upstream's error says about whether a method exists on that upstream. It is three-valued on purpose: "no opinion" has to be distinguishable from "definitely absent", so that a transient failure is never mistaken for a missing method. 

const (
	// MethodAvailabilityUnknown - the error says nothing about the method's existence.
	MethodAvailabilityUnknown MethodAvailability = iota
	// MethodNotAvailable - the upstream does not have this method.
	MethodNotAvailable
	// MethodAvailable - the upstream has the method and rejected this particular call,
	// typically because the params were wrong.
	MethodAvailable
)

func ClassifyMethodAvailability 

func ClassifyMethodAvailability(respError *ResponseError) MethodAvailability

ClassifyMethodAvailability decides what an upstream error says about whether a method exists. Not-available patterns are checked before the params ones: an unavailable method is the answer that matters, and a params complaint only ever means the method is there.

type ParsedEvent 

type ParsedEvent interface {
	Raw() []byte
}

ParsedEvent is the pre-parsed view of a WsResponse Message. See WsResponse.ParsedEvent.

type RealIdHolder 

type RealIdHolder interface {
	// RealId is the client's JSON-RPC id as text, string ids unquoted.
	RealId() string
}

RealIdHolder is a request that carries the client's own JSON-RPC id, as opposed to Id(), nodecore's internal tag. Only JSON-RPC requests have one; consumers that need it (quorum header checks, go-jsonrpc channel framing) type-assert for it.

type RemoveUpstreamEvent 

type RemoveUpstreamEvent struct{}

type ReplyError 

type ReplyError struct {
	ErrorKind ResponseErrorKind
	// contains filtered or unexported fields
}

func NewPartialFailure 

func NewPartialFailure(request RequestHolder, responseError *ResponseError) *ReplyError

func NewReplyError 

func NewReplyError(id string, responseError *ResponseError, responseType RequestType, errorKind ResponseErrorKind) *ReplyError

func NewTotalFailure 

func NewTotalFailure(request RequestHolder, responseError *ResponseError) *ReplyError

func NewTotalFailureFromErr 

func NewTotalFailureFromErr(id string, err error, responseType RequestType) *ReplyError

func (*ReplyError) EncodeResponse 

func (r *ReplyError) EncodeResponse(realId []byte) io.Reader

func (*ReplyError) GetError 

func (r *ReplyError) GetError() *ResponseError

func (ReplyError) GetStreamHint 

func (ReplyError) GetStreamHint() StreamHint

func (*ReplyError) HasError 

func (r *ReplyError) HasError() bool

func (*ReplyError) HasStream 

func (r *ReplyError) HasStream() bool

func (*ReplyError) Id 

func (r *ReplyError) Id() string

func (*ReplyError) ResponseCode 

func (r *ReplyError) ResponseCode() int

func (*ReplyError) ResponseHeaders 

func (r *ReplyError) ResponseHeaders() http.Header

func (*ReplyError) ResponseResult 

func (r *ReplyError) ResponseResult() []byte

func (*ReplyError) ResponseResultString 

func (r *ReplyError) ResponseResultString() (string, error)

func (*ReplyError) ResponseTrailers 

func (r *ReplyError) ResponseTrailers() map[string][]string

func (*ReplyError) WithResponseHeaders 

func (r *ReplyError) WithResponseHeaders(headers http.Header) *ReplyError

func (*ReplyError) WithResponseTrailers 

func (r *ReplyError) WithResponseTrailers(trailers map[string][]string) *ReplyError

type RequestAndSelector 

type RequestAndSelector struct {
	Children []RequestSelector
}

func (RequestAndSelector) Key 

func (s RequestAndSelector) Key() string

type RequestAnySelector 

type RequestAnySelector struct{}

func (RequestAnySelector) Key 

func (RequestAnySelector) Key() string

type RequestBlockTag 

type RequestBlockTag int
const (
	BlockTagLatest RequestBlockTag = iota
	BlockTagSafe
	BlockTagFinalized
)

type RequestBlockTagSelector 

type RequestBlockTagSelector struct {
	Tag RequestBlockTag
}

func (RequestBlockTagSelector) Key 

func (s RequestBlockTagSelector) Key() string

type RequestExistsSelector 

type RequestExistsSelector struct {
	Name string
}

func (RequestExistsSelector) Key 

func (s RequestExistsSelector) Key() string

type RequestHeightSelector 

type RequestHeightSelector struct {
	Height int64
}

func (RequestHeightSelector) Key 

func (s RequestHeightSelector) Key() string

type RequestHolder 

type RequestHolder interface {
	Id() string
	Method() string
	RequestParams() *RequestParams
	Body() ([]byte, error)
	ParseParams(ctx context.Context) specs.MethodParam
	RequestType() RequestType
	RequestHash() string
	SpecMethod() *specs.Method
	RequestObserver() *RequestObserver
	Selectors() []RequestSelector

	ModifyParams(ctx context.Context, newValue any)

	IsStream() bool
	IsSubscribe() bool
}

type RequestKind 

type RequestKind int
const (
	UnknownReqKind RequestKind = iota
	InternalUnary
	InternalSubscription
	Local
	Cached
	Unary
	Subscription
)

func (RequestKind) String 

func (r RequestKind) String() string

type RequestLabelSelector 

type RequestLabelSelector struct {
	Name   string
	Values []string
}

func (RequestLabelSelector) Key 

func (s RequestLabelSelector) Key() string

type RequestLowerHeightSelector 

type RequestLowerHeightSelector struct {
	Height         int64
	LowerBoundType LowerBoundType
	TimeOffset     int64
	HeightDelta    int64
}

func (RequestLowerHeightSelector) Key 

func (s RequestLowerHeightSelector) Key() string

type RequestNotSelector 

type RequestNotSelector struct {
	Child RequestSelector
}

func (RequestNotSelector) Key 

func (s RequestNotSelector) Key() string

type RequestObserver 

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

func NewRequestObserver 

func NewRequestObserver(isSub bool) *RequestObserver

func (*RequestObserver) AddResult 

func (b *RequestObserver) AddResult(result RequestResult, final bool)

func (*RequestObserver) GetChain 

func (b *RequestObserver) GetChain() chains.Chain

func (*RequestObserver) GetRequestKind 

func (b *RequestObserver) GetRequestKind() RequestKind

func (*RequestObserver) GetResults 

func (b *RequestObserver) GetResults() []RequestResult

func (*RequestObserver) TrackUpstreamCall 

func (b *RequestObserver) TrackUpstreamCall() func()

func (*RequestObserver) WithApiKey 

func (b *RequestObserver) WithApiKey(apiKey string) *RequestObserver

func (*RequestObserver) WithChain 

func (b *RequestObserver) WithChain(chain chains.Chain) *RequestObserver

func (*RequestObserver) WithMethod 

func (b *RequestObserver) WithMethod(method string) *RequestObserver

func (*RequestObserver) WithRequestKind 

func (b *RequestObserver) WithRequestKind(kind RequestKind) *RequestObserver

type RequestOrSelector 

type RequestOrSelector struct {
	Children []RequestSelector
}

func (RequestOrSelector) Key 

func (s RequestOrSelector) Key() string

type RequestParams 

type RequestParams struct {
	Headers     map[string][]string
	QueryParams map[string][]string
	PathParams  []string
}

RequestParams carries the non-body components of a REST request as the proxy sees them. Headers and QueryParams use []string values so that repeated keys (e.g. "?ids=a&ids=b" or a header appearing twice) survive the round-trip - the upstream sees them just like the client sent them.

PathParams holds the wildcard captures from the spec's URL template in path order. For HTTP requests they come out of PathMatcher; for gRPC NativeCall they come straight from the protobuf RestData.path_params field.

type RequestResult 

type RequestResult interface {
	// contains filtered or unexported methods
}

type RequestSelector 

type RequestSelector interface {
	Key() string
	// contains filtered or unexported methods
}

RequestSelector is a sealed protocol-neutral selector sum type. Keep concrete selector payloads separate so invalid field combinations (for example both a block tag and numeric height) cannot be represented by construction.

type RequestSlotHeightSelector 

type RequestSlotHeightSelector struct {
	SlotHeight int64
}

func (RequestSlotHeightSelector) Key 

func (s RequestSlotHeightSelector) Key() string

type RequestType 

type RequestType int
const (
	Rest RequestType = iota
	JsonRpc
	Ws
	Grpc
	Unknown
)

func (RequestType) CanBeServedBy 

func (r RequestType) CanBeServedBy(connectorTypes []specs.ApiConnectorType) bool

CanBeServedBy reports whether a request of this wire shape can be sent through at least one of the given connector types. The only boundary policed is gRPC: proto frames and JSON bodies are different codecs and nothing translates between them, so a gRPC request needs a gRPC-family connector (grpc or grpc-additional) and a non-gRPC request needs any other connector. Within the HTTP family the shapes are deliberately not checked - one connector may serve two forms (tendermint), a spec may declare one form and ship another, and method translators rewrite a JSON-RPC method into a REST call right before the connector.

func (RequestType) String 

func (r RequestType) String() string

type RequestUnsupportedSelector 

type RequestUnsupportedSelector struct {
	Reason string
}

func (RequestUnsupportedSelector) Key 

func (s RequestUnsupportedSelector) Key() string

type ResponseError 

type ResponseError struct {
	Code    int
	Message string
	Data    interface{}
}

func AuthError 

func AuthError(cause error) *ResponseError

func ClientError 

func ClientError(cause error) *ResponseError

func CtxError 

func CtxError(cause error) *ResponseError

func IncorrectResponseBodyError 

func IncorrectResponseBodyError(cause error) *ResponseError

func InvalidParamsError 

func InvalidParamsError(message string) *ResponseError

func MethodNotAvailableOverTransportError 

func MethodNotAvailableOverTransportError(method string, requestType RequestType) *ResponseError

MethodNotAvailableOverTransportError answers a method that exists on the chain but is bound to connector types that cannot carry this request's wire shape (a gRPC method called over JSON-RPC, for example). It shares the code of NotSupportedMethodError so the observer and error classification keep treating it as a client-side rejection rather than an upstream failure.

func NewGrpcStatusError 

func NewGrpcStatusError(code codes.Code, message string) *ResponseError

NewGrpcStatusError builds a ResponseError carrying a gRPC status as its typed GrpcStatus, so the ingress replays it to the client verbatim instead of degrading it to INTERNAL (e.g. receive-side errors: client cancel, oversized request message).

func NewGrpcStatusResponseError 

func NewGrpcStatusResponseError(grpcStatus *GrpcStatus) *ResponseError

NewGrpcStatusResponseError builds the ResponseError that carries an upstream gRPC status verbatim. The int code is informational; routing and rendering read the typed GrpcStatus in Data.

func NoApiConnectorsError 

func NoApiConnectorsError(method string) *ResponseError

func NoAvailableUpstreamsError 

func NoAvailableUpstreamsError() *ResponseError

func NoAvailableUpstreamsErrorWithCause 

func NoAvailableUpstreamsErrorWithCause(cause string) *ResponseError

func NotSupportedMethodError 

func NotSupportedMethodError(method string) *ResponseError

func ParseError 

func ParseError() *ResponseError

func QuorumInsufficientSignaturesError 

func QuorumInsufficientSignaturesError(got, required int) *ResponseError

func QuorumInvalidSignatureError 

func QuorumInvalidSignatureError(providerID string) *ResponseError

func QuorumMalformedHeaderError 

func QuorumMalformedHeaderError(cause error) *ResponseError

func QuorumMissingSignaturesError 

func QuorumMissingSignaturesError() *ResponseError

func QuorumNotSupportedError 

func QuorumNotSupportedError(reason string) *ResponseError

func QuorumUnexpectedRequestIDError 

func QuorumUnexpectedRequestIDError(expected, got string) *ResponseError

func QuorumUnknownProviderError 

func QuorumUnknownProviderError(providerID string) *ResponseError

func QuorumVerificationError 

func QuorumVerificationError(cause error) *ResponseError

func RateLimitError 

func RateLimitError() *ResponseError

func RequestTimeoutError 

func RequestTimeoutError() *ResponseError

func ResponseErrorWithData 

func ResponseErrorWithData(code int, message string, data interface{}) *ResponseError

func ResponseErrorWithMessage 

func ResponseErrorWithMessage(message string) *ResponseError

func ServerError 

func ServerError() *ResponseError

func ServerErrorWithCause 

func ServerErrorWithCause(cause error) *ResponseError

func SubscribeTotalFailureError 

func SubscribeTotalFailureError() *ResponseError

func SubscriberTooSlowError 

func SubscriberTooSlowError() *ResponseError

SubscriberTooSlowError is the terminal error delivered to a subscriber that could not keep up with the shared source's fan-out (its buffer overflowed). It uses the SubscribeTotalFailure code so the client treats it like any other terminal failure and can resubscribe, but carries a distinct message so the cause is visible in logs.

func UnsupportedBlockTagError 

func UnsupportedBlockTagError(chain, tag string) *ResponseError

func WrongChainError 

func WrongChainError(chain string) *ResponseError

func (*ResponseError) Error 

func (b *ResponseError) Error() string

type ResponseErrorKind 

type ResponseErrorKind int
const (
	PartialFailure ResponseErrorKind = iota
	TotalFailure
)

type ResponseHolder 

type ResponseHolder interface {
	ResponseResult() []byte
	ResponseResultString() (string, error)
	ResponseCode() int
	GetError() *ResponseError
	EncodeResponse(realId []byte) io.Reader
	HasError() bool
	HasStream() bool
	// GetStreamHint returns the streaming hint located by the connector's
	// single-pass first-chunk analysis, or nil when the response carries no
	// hint (non-streaming, REST, errors, subscriptions). Only the gRPC
	// result-unwrap consumer uses it.
	GetStreamHint() StreamHint
	Id() string
}

func NewGrpcUpstreamErrorResponse 

func NewGrpcUpstreamErrorResponse(request RequestHolder, grpcStatus *GrpcStatus) ResponseHolder

NewGrpcUpstreamErrorResponse maps a non-OK upstream status onto the response shape that routes it correctly. Classification drives routing; the original status rides through verbatim inside the ResponseError:

  • client errors (CANCELLED, INVALID_ARGUMENT, NOT_FOUND, ALREADY_EXISTS, FAILED_PRECONDITION, OUT_OF_RANGE) and upstream auth/config problems (PERMISSION_DENIED, UNAUTHENTICATED) pass through with no retry;
  • transient upstream failures (UNKNOWN, INTERNAL, DATA_LOSS, UNAVAILABLE, ABORTED), timeouts (DEADLINE_EXCEEDED) and throttling (RESOURCE_EXHAUSTED) become partial failures, i.e. retryable on another upstream;
  • UNIMPLEMENTED is a partial failure too, and additionally bans the method on the upstream via ClassifyMethodAvailability.

type ResponseHolderWrapper 

type ResponseHolderWrapper struct {
	UpstreamId            string
	RequestId             string
	UpstreamNodeVersion   string
	FinalizationBlockType *BlockType
	FinalizationBlock     Block
	Response              ResponseHolder
}

type ResponseKind 

type ResponseKind int
const (
	UnknownRespKind ResponseKind = iota
	Ok
	Cancelled
	Error
	RetryableError
	RoutingError
)

func GetRespKindFromResponse 

func GetRespKindFromResponse(response ResponseHolder) ResponseKind

func (ResponseKind) String 

func (r ResponseKind) String() string

type ResponseReceivedHook 

type ResponseReceivedHook interface {
	OnResponseReceived(ctx context.Context, request RequestHolder, respWrapper *ResponseHolderWrapper)
}

type ResultCounter 

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

ResultCounter is a byte-level state machine that tracks how much of a JSON value's structure is still open.

For object/array kinds, the counter ignores brackets inside string literals and respects '\\' escapes. For string kind, it tracks escapes. For scalar kind (null / true / false / number), it terminates on the first byte that cannot belong to the scalar (',', '}', ']', or whitespace).

func (*ResultCounter) Step 

func (c *ResultCounter) Step(b byte) StepResult

Step feeds one byte into the counter and returns what the caller should do with that byte.

type ResultType 

type ResultType int
const (
	ResultOk ResultType = iota
	ResultOkWithError
	ResultPartialFailure
	ResultTotalFailure
	ResultStop
)

func GetResponseType 

func GetResponseType(wrapper *ResponseHolderWrapper, err error) ResultType

type StateUpstreamEvent 

type StateUpstreamEvent struct {
	State *UpstreamState
}

type StatusUpstreamStateEvent 

type StatusUpstreamStateEvent struct {
	// Status is the base availability reported by a health probe (setStatus).
	// It is ignored when Lag != nil.
	Status AvailabilityStatus
	// Lag is a discriminator, not the source of truth: non-nil marks a head-lag
	// update (setLag) that only asks the event loop to re-derive availability from
	// the upstream's current lag (stored on the upstream itself); nil
	// marks a setStatus that updates the base availability. A pointer is used so
	// this stays distinguishable from a setStatus even when the lag is 0.
	Lag *int64
}

func (*StatusUpstreamStateEvent) ProcessEvent 

func (s *StatusUpstreamStateEvent) ProcessEvent(state UpstreamState) UpstreamState

ProcessEvent is a no-op: the setStatus/setLag mutation and the derived effective availability are handled by processStateEvents, which owns both the base-availability tracking and the per-chain syncing threshold. It exists only to satisfy AbstractUpstreamStateEvent.

func (*StatusUpstreamStateEvent) Same 

func (s *StatusUpstreamStateEvent) Same(_ UpstreamState) bool

type StepResult 

type StepResult uint8

StepResult tells the caller what to do with the byte just fed to Step. 

const (
	// StepContinue byte is part of the result; keep reading.
	StepContinue StepResult = iota
	// StepFinishHere byte is the last byte of the result (e.g. closing
	// '}', ']', or '"'); include it, then stop.
	StepFinishHere
	// StepStopBefore byte is NOT part of the result (scalar delimiter);
	// do not include it; stop.
	StepStopBefore
)

type StopRetryErr 

type StopRetryErr struct {
}

func (StopRetryErr) Error 

func (s StopRetryErr) Error() string

type StreamHint 

type StreamHint interface {
	Type() RequestType
}

StreamHint is a marker for the analysis attached to a streaming response, produced by the connector's first-chunk pass and consumed by the matching streaming path. Type reports which response shape the hint describes; a consumer asserts to the concrete hint of that type to read its fields. A nil StreamHint means no hint is available (a non-streaming or REST response).

type SubResponse 

type SubResponse interface {
	GetMessage() []byte
	GetError() *ResponseError
	GetUpstreamId() string
	GetParsedEvent() ParsedEvent
	IsEnd() bool
}

SubResponse is one event of an upstream subscription/stream: a data notification (GetMessage), a terminal error (GetError), or a clean end of the stream (IsEnd - a bounded gRPC stream completing; it may carry trailers). Implementations are transport-specific; consumers use only these accessors. Producers must stamp every event with the originating upstream id (GetUpstreamId) - there is no setter, and consumers forward events as-is.

type SubscribeConnectorState 

type SubscribeConnectorState int
const (
	WsDisconnected SubscribeConnectorState = iota
	WsConnected
)

type SubscriptionEndResponse 

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

SubscriptionEndResponse is the final frame of a stream that completed cleanly (a bounded gRPC stream). It carries no payload - only the upstream trailers the ingress must deliver with the OK status.

func NewSubscriptionEndResponse 

func NewSubscriptionEndResponse(id string) *SubscriptionEndResponse

func (*SubscriptionEndResponse) EncodeResponse 

func (s *SubscriptionEndResponse) EncodeResponse(_ []byte) io.Reader

func (*SubscriptionEndResponse) GetError 

func (s *SubscriptionEndResponse) GetError() *ResponseError

func (SubscriptionEndResponse) GetStreamHint 

func (SubscriptionEndResponse) GetStreamHint() StreamHint

func (*SubscriptionEndResponse) HasError 

func (s *SubscriptionEndResponse) HasError() bool

func (*SubscriptionEndResponse) HasStream 

func (s *SubscriptionEndResponse) HasStream() bool

func (*SubscriptionEndResponse) Id 

func (s *SubscriptionEndResponse) Id() string

func (*SubscriptionEndResponse) IsEnd 

func (s *SubscriptionEndResponse) IsEnd() bool

func (*SubscriptionEndResponse) ResponseCode 

func (s *SubscriptionEndResponse) ResponseCode() int

func (*SubscriptionEndResponse) ResponseHeaders 

func (s *SubscriptionEndResponse) ResponseHeaders() http.Header

func (*SubscriptionEndResponse) ResponseResult 

func (s *SubscriptionEndResponse) ResponseResult() []byte

func (*SubscriptionEndResponse) ResponseResultString 

func (s *SubscriptionEndResponse) ResponseResultString() (string, error)

func (*SubscriptionEndResponse) ResponseTrailers 

func (s *SubscriptionEndResponse) ResponseTrailers() map[string][]string

func (*SubscriptionEndResponse) WithResponseHeaders 

func (s *SubscriptionEndResponse) WithResponseHeaders(headers http.Header) *SubscriptionEndResponse

func (*SubscriptionEndResponse) WithResponseTrailers 

func (s *SubscriptionEndResponse) WithResponseTrailers(trailers map[string][]string) *SubscriptionEndResponse

type SubscriptionEventResponse 

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

SubscriptionEventResponse is one event of a subscription/stream as the client receives it. The encoder decides the presentation; the transport metadata (gRPC headers/trailers) rides along for the gRPC ingress.

func NewChannelSubscriptionEventResponse 

func NewChannelSubscriptionEventResponse(id, method string, payload []byte, channelId uint64) *SubscriptionEventResponse

NewChannelSubscriptionEventResponse is the event in the go-jsonrpc channel notification shape used by celestia-node clients: {"jsonrpc":"2.0","method":<method>,"params":[<channelId>,<payload>]}. With no payload the params hold the channel id alone, which is how a channel close (xrpc.ch.close) is announced.

func NewJsonRpcSubscriptionEventResponse 

func NewJsonRpcSubscriptionEventResponse(id, method string, payload []byte, subId json.RawMessage) *SubscriptionEventResponse

NewJsonRpcSubscriptionEventResponse is the event wrapped in a JSON-RPC notification envelope referencing the client subscription id (the WS server).

func NewSubscriptionEventResponse 

func NewSubscriptionEventResponse(id string, payload []byte) *SubscriptionEventResponse

NewSubscriptionEventResponse is the bare-payload event (result-only consumers: the gRPC ingress, the emerald server).

func (*SubscriptionEventResponse) EncodeResponse 

func (s *SubscriptionEventResponse) EncodeResponse(_ []byte) io.Reader

EncodeResponse ignores realId: an event references the subscription id, not the request id.

func (*SubscriptionEventResponse) GetError 

func (s *SubscriptionEventResponse) GetError() *ResponseError

func (SubscriptionEventResponse) GetStreamHint 

func (SubscriptionEventResponse) GetStreamHint() StreamHint

func (*SubscriptionEventResponse) HasError 

func (s *SubscriptionEventResponse) HasError() bool

func (*SubscriptionEventResponse) HasStream 

func (s *SubscriptionEventResponse) HasStream() bool

func (*SubscriptionEventResponse) Id 

func (s *SubscriptionEventResponse) Id() string

func (*SubscriptionEventResponse) IsEnd 

func (s *SubscriptionEventResponse) IsEnd() bool

func (*SubscriptionEventResponse) ResponseCode 

func (s *SubscriptionEventResponse) ResponseCode() int

func (*SubscriptionEventResponse) ResponseHeaders 

func (s *SubscriptionEventResponse) ResponseHeaders() http.Header

func (*SubscriptionEventResponse) ResponseResult 

func (s *SubscriptionEventResponse) ResponseResult() []byte

func (*SubscriptionEventResponse) ResponseResultString 

func (s *SubscriptionEventResponse) ResponseResultString() (string, error)

func (*SubscriptionEventResponse) ResponseTrailers 

func (s *SubscriptionEventResponse) ResponseTrailers() map[string][]string

func (*SubscriptionEventResponse) WithResponseHeaders 

func (s *SubscriptionEventResponse) WithResponseHeaders(headers http.Header) *SubscriptionEventResponse

func (*SubscriptionEventResponse) WithResponseTrailers 

func (s *SubscriptionEventResponse) WithResponseTrailers(trailers map[string][]string) *SubscriptionEventResponse

type SubscriptionResponseHolder 

type SubscriptionResponseHolder interface {
	ResponseHolder
	// IsEnd reports the final frame of a stream that completed cleanly. It
	// carries no payload, only the trailers the upstream closed with.
	IsEnd() bool
}

SubscriptionResponseHolder is a client-facing frame of a subscription or stream: an event carrying a payload, or the clean end of a bounded stream.

type UnaryRequestResult 

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

func NewUnaryRequestResult 

func NewUnaryRequestResult() *UnaryRequestResult

func (*UnaryRequestResult) GetApiKey 

func (u *UnaryRequestResult) GetApiKey() string

func (*UnaryRequestResult) GetChain 

func (u *UnaryRequestResult) GetChain() chains.Chain

func (*UnaryRequestResult) GetDuration 

func (u *UnaryRequestResult) GetDuration() float64

func (*UnaryRequestResult) GetMethod 

func (u *UnaryRequestResult) GetMethod() string

func (*UnaryRequestResult) GetReqKind 

func (u *UnaryRequestResult) GetReqKind() RequestKind

func (*UnaryRequestResult) GetRespKind 

func (u *UnaryRequestResult) GetRespKind() ResponseKind

func (*UnaryRequestResult) GetTimestamp 

func (u *UnaryRequestResult) GetTimestamp() time.Time

func (*UnaryRequestResult) GetUpstreamId 

func (u *UnaryRequestResult) GetUpstreamId() string

func (*UnaryRequestResult) IsSuccessfulRetry 

func (u *UnaryRequestResult) IsSuccessfulRetry() bool

func (*UnaryRequestResult) WithDuration 

func (u *UnaryRequestResult) WithDuration(duration float64) *UnaryRequestResult

func (*UnaryRequestResult) WithRespKindFromResponse 

func (u *UnaryRequestResult) WithRespKindFromResponse(response ResponseHolder) *UnaryRequestResult

func (*UnaryRequestResult) WithSuccessfulRetry 

func (u *UnaryRequestResult) WithSuccessfulRetry() *UnaryRequestResult

func (*UnaryRequestResult) WithUpstreamId 

func (u *UnaryRequestResult) WithUpstreamId(upstreamId string) *UnaryRequestResult

type UnbanMethodUpstreamStateEvent 

type UnbanMethodUpstreamStateEvent struct {
	Method string
}

func (*UnbanMethodUpstreamStateEvent) ProcessEvent 

func (u *UnbanMethodUpstreamStateEvent) ProcessEvent(state UpstreamState) UpstreamState

func (*UnbanMethodUpstreamStateEvent) Same 

func (u *UnbanMethodUpstreamStateEvent) Same(_ UpstreamState) bool

type UnsupportedMethodsUpstreamStateEvent 

type UnsupportedMethodsUpstreamStateEvent struct {
	Methods mapset.Set[string]
}

UnsupportedMethodsUpstreamStateEvent replaces the upstream's detected-unsupported set wholesale. Detection produces a whole-set view - the pipeline has already merged every detector's verdict - so this carries the full set rather than incremental strip/restore. It is deliberately not BanMethodUpstreamStateEvent: a ban expires after Methods.BanDuration, whereas a method the node was never built with must stay gone until detection itself says otherwise.

func (*UnsupportedMethodsUpstreamStateEvent) ProcessEvent 

func (u *UnsupportedMethodsUpstreamStateEvent) ProcessEvent(state UpstreamState) UpstreamState

ProcessEvent is a no-op. The detected set is combined with the ban set and the upstream's config by processStateEvents, which owns both.

func (*UnsupportedMethodsUpstreamStateEvent) Same 

func (u *UnsupportedMethodsUpstreamStateEvent) Same(_ UpstreamState) bool

Same always reports false: the comparison needs the previously detected set, which processStateEvents tracks rather than UpstreamState, so dedup happens there - the same arrangement the ban events use.

type UpstreamEvent 

type UpstreamEvent struct {
	Id        string
	Chain     chains.Chain
	EventType UpstreamEventType
}

type UpstreamEventType 

type UpstreamEventType interface {
	// contains filtered or unexported methods
}

type UpstreamGrpcRequest 

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

UpstreamGrpcRequest is a gRPC call (unary or server-streaming - the spec decides, see IsSubscribe) as the proxy sees it: the full method string ("/sui.rpc.v2.LedgerService/GetObject"), the serialized request message exactly as the client produced it (the 5-byte gRPC wire frame prefix never appears at this layer - grpc-go adds and strips it below the codec), and the client metadata in RequestParams.Headers. The body is never parsed - the traffic path is bytes-only.

func NewInternalUpstreamGrpcRequest 

func NewInternalUpstreamGrpcRequest(method string, body []byte, chain chains.Chain) *UpstreamGrpcRequest

NewInternalUpstreamGrpcRequest builds an internally-originated gRPC request for chain-specific probes: the caller proto.Marshals a typed request into body and the connector sends it verbatim.

func NewUpstreamGrpcRequest 

func NewUpstreamGrpcRequest(id, method string, requestParams *RequestParams, body []byte, specName string, selectors ...RequestSelector) *UpstreamGrpcRequest

func (*UpstreamGrpcRequest) Body 

func (u *UpstreamGrpcRequest) Body() ([]byte, error)

func (*UpstreamGrpcRequest) Id 

func (u *UpstreamGrpcRequest) Id() string

func (*UpstreamGrpcRequest) IsStream 

func (u *UpstreamGrpcRequest) IsStream() bool

func (*UpstreamGrpcRequest) IsSubscribe 

func (u *UpstreamGrpcRequest) IsSubscribe() bool

IsSubscribe follows the spec: both gRPC server-stream call types are routed as subscriptions.

func (*UpstreamGrpcRequest) Method 

func (u *UpstreamGrpcRequest) Method() string

func (*UpstreamGrpcRequest) ModifyParams 

func (u *UpstreamGrpcRequest) ModifyParams(_ context.Context, _ any)

func (*UpstreamGrpcRequest) ParseParams 

func (u *UpstreamGrpcRequest) ParseParams(_ context.Context) specs.MethodParam

func (*UpstreamGrpcRequest) RequestHash 

func (u *UpstreamGrpcRequest) RequestHash() string

RequestHash reuses calculateRestHash's injective tag+length framing over exactly the sections a gRPC call has: method, body, and the selector label key. There are no path/query sections, and method shapes ("/package.Service/Method" vs "VERB#/path") keep the two families apart.

func (*UpstreamGrpcRequest) RequestObserver 

func (u *UpstreamGrpcRequest) RequestObserver() *RequestObserver

func (*UpstreamGrpcRequest) RequestParams 

func (u *UpstreamGrpcRequest) RequestParams() *RequestParams

func (*UpstreamGrpcRequest) RequestType 

func (u *UpstreamGrpcRequest) RequestType() RequestType

func (*UpstreamGrpcRequest) Selectors 

func (u *UpstreamGrpcRequest) Selectors() []RequestSelector

func (*UpstreamGrpcRequest) SpecMethod 

func (u *UpstreamGrpcRequest) SpecMethod() *specs.Method

type UpstreamJsonRpcRequest 

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

func NewInternalSubUpstreamJsonRpcRequest 

func NewInternalSubUpstreamJsonRpcRequest(method string, params any, chain chains.Chain) (*UpstreamJsonRpcRequest, error)

func NewInternalUpstreamJsonRpcRequest 

func NewInternalUpstreamJsonRpcRequest(method string, params any, chain chains.Chain) (*UpstreamJsonRpcRequest, error)

func NewStreamUpstreamJsonRpcRequest 

func NewStreamUpstreamJsonRpcRequest(id string, jsonRpcRequest JsonRpcRequestBody, specName string, selectors ...RequestSelector) *UpstreamJsonRpcRequest

func NewUpstreamJsonRpcRequest 

func NewUpstreamJsonRpcRequest(id string, jsonRpcRequest JsonRpcRequestBody, isSub bool, specName string, selectors ...RequestSelector) *UpstreamJsonRpcRequest

func NewUpstreamJsonRpcRequestWithSpecMethod 

func NewUpstreamJsonRpcRequestWithSpecMethod(method string, params any, specMethod *specs.Method) (*UpstreamJsonRpcRequest, error)

func (*UpstreamJsonRpcRequest) Body 

func (u *UpstreamJsonRpcRequest) Body() ([]byte, error)

func (*UpstreamJsonRpcRequest) Headers 

func (u *UpstreamJsonRpcRequest) Headers() map[string]string

func (*UpstreamJsonRpcRequest) Id 

func (u *UpstreamJsonRpcRequest) Id() string

func (*UpstreamJsonRpcRequest) IsStream 

func (u *UpstreamJsonRpcRequest) IsStream() bool

func (*UpstreamJsonRpcRequest) IsSubscribe 

func (u *UpstreamJsonRpcRequest) IsSubscribe() bool

func (*UpstreamJsonRpcRequest) Method 

func (u *UpstreamJsonRpcRequest) Method() string

func (*UpstreamJsonRpcRequest) ModifyParams 

func (u *UpstreamJsonRpcRequest) ModifyParams(ctx context.Context, newValue any)

func (*UpstreamJsonRpcRequest) ParseParams 

func (u *UpstreamJsonRpcRequest) ParseParams(ctx context.Context) specs.MethodParam

func (*UpstreamJsonRpcRequest) RealId 

func (u *UpstreamJsonRpcRequest) RealId() string

RealId returns the textual form of the JSON-RPC id actually placed on the wire (u.realId), unwrapped from the outer JSON quotes for string ids. This is what upstreams echo back in correlation headers (e.g. the QR quorum signature headers), and it differs from Id() which is nodecore's internal UUID tag used to route responses.

func (*UpstreamJsonRpcRequest) RequestHash 

func (u *UpstreamJsonRpcRequest) RequestHash() string

func (*UpstreamJsonRpcRequest) RequestObserver 

func (u *UpstreamJsonRpcRequest) RequestObserver() *RequestObserver

func (*UpstreamJsonRpcRequest) RequestParams 

func (u *UpstreamJsonRpcRequest) RequestParams() *RequestParams

func (*UpstreamJsonRpcRequest) RequestType 

func (u *UpstreamJsonRpcRequest) RequestType() RequestType

func (*UpstreamJsonRpcRequest) Selectors 

func (u *UpstreamJsonRpcRequest) Selectors() []RequestSelector

func (*UpstreamJsonRpcRequest) SpecMethod 

func (u *UpstreamJsonRpcRequest) SpecMethod() *specs.Method

type UpstreamRestRequest 

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

func NewInternalUpstreamRestRequest 

func NewInternalUpstreamRestRequest(methodTemplate string, requestParams *RequestParams, chain chains.Chain) *UpstreamRestRequest

NewInternalUpstreamRestRequest builds an internally-originated REST request from a spec method template ("GET#/v2/status", "GET#/v2/blocks/*") and the matching RequestParams. Wildcard captures live on requestParams.PathParams and query on requestParams.QueryParams - the connector rebuilds the literal upstream URL from template + captures via utils.BuildRestURL. This mirrors NewUpstreamRestRequest so internal probes carry the bounded template as their Method() (not a per-resource literal path) and resolve a real spec method.

func NewInternalUpstreamRestRequestWithBody 

func NewInternalUpstreamRestRequestWithBody(methodTemplate string, requestParams *RequestParams, body []byte, chain chains.Chain) *UpstreamRestRequest

func NewStreamUpstreamRestRequest 

func NewStreamUpstreamRestRequest(id, methodTemplate string, requestParams *RequestParams, body []byte, specName string, selectors ...RequestSelector) *UpstreamRestRequest

func NewUpstreamRestRequest 

func NewUpstreamRestRequest(id, methodTemplate string, requestParams *RequestParams, body []byte, specName string, selectors ...RequestSelector) *UpstreamRestRequest

func (*UpstreamRestRequest) Body 

func (u *UpstreamRestRequest) Body() ([]byte, error)

func (*UpstreamRestRequest) Id 

func (u *UpstreamRestRequest) Id() string

func (*UpstreamRestRequest) IsStream 

func (u *UpstreamRestRequest) IsStream() bool

func (*UpstreamRestRequest) IsSubscribe 

func (u *UpstreamRestRequest) IsSubscribe() bool

func (*UpstreamRestRequest) Method 

func (u *UpstreamRestRequest) Method() string

func (*UpstreamRestRequest) ModifyParams 

func (u *UpstreamRestRequest) ModifyParams(_ context.Context, _ any)

func (*UpstreamRestRequest) ParseParams 

func (u *UpstreamRestRequest) ParseParams(_ context.Context) specs.MethodParam

func (*UpstreamRestRequest) RequestHash 

func (u *UpstreamRestRequest) RequestHash() string

func (*UpstreamRestRequest) RequestObserver 

func (u *UpstreamRestRequest) RequestObserver() *RequestObserver

func (*UpstreamRestRequest) RequestParams 

func (u *UpstreamRestRequest) RequestParams() *RequestParams

func (*UpstreamRestRequest) RequestType 

func (u *UpstreamRestRequest) RequestType() RequestType

func (*UpstreamRestRequest) Selectors 

func (u *UpstreamRestRequest) Selectors() []RequestSelector

func (*UpstreamRestRequest) SpecMethod 

func (u *UpstreamRestRequest) SpecMethod() *specs.Method

type UpstreamState 

type UpstreamState struct {
	Status          AvailabilityStatus
	HeadData        Block
	UpstreamMethods methods.Methods
	Caps            mapset.Set[Cap]
	UpstreamIndex   string

	RateLimiterBudget   *ratelimiter.RateLimitBudget
	AutoTuneRateLimiter *ratelimiter.UpstreamAutoTune

	BlockInfo       *BlockInfo
	LowerBoundsInfo *LowerBoundInfo
	Labels          *Labels
}

func DefaultUpstreamState 

func DefaultUpstreamState(upstreamMethods methods.Methods, caps mapset.Set[Cap], upstreamIndex string, rt *ratelimiter.RateLimitBudget, autoTuneRateLimiter *ratelimiter.UpstreamAutoTune) UpstreamState

type UpstreamSubscriptionResponse 

type UpstreamSubscriptionResponse interface {
	ResponseChan() chan SubResponse
	OpId() string
}

type ValidUpstreamEvent 

type ValidUpstreamEvent struct {
	State *UpstreamState
}

ValidUpstreamEvent carries the upstream-state snapshot taken at the moment the settings validators declared the upstream healthy again. The chain supervisor uses it to re-register the upstream immediately; without the snapshot the re-add would have to wait for a later StateUpstreamEvent, which only fires when some sub-state actually changes.

type ValidUpstreamStateEvent 

type ValidUpstreamStateEvent struct{}

func (*ValidUpstreamStateEvent) ProcessEvent 

func (v *ValidUpstreamStateEvent) ProcessEvent(state UpstreamState) UpstreamState

func (*ValidUpstreamStateEvent) Same 

func (v *ValidUpstreamStateEvent) Same(_ UpstreamState) bool

type WsJsonRpcResponse 

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

func NewWsJsonRpcResponse 

func NewWsJsonRpcResponse(id string, result []byte, error *ResponseError) *WsJsonRpcResponse

func (*WsJsonRpcResponse) EncodeResponse 

func (w *WsJsonRpcResponse) EncodeResponse(realId []byte) io.Reader

func (*WsJsonRpcResponse) GetError 

func (w *WsJsonRpcResponse) GetError() *ResponseError

func (WsJsonRpcResponse) GetStreamHint 

func (WsJsonRpcResponse) GetStreamHint() StreamHint

func (*WsJsonRpcResponse) HasError 

func (w *WsJsonRpcResponse) HasError() bool

func (*WsJsonRpcResponse) HasStream 

func (w *WsJsonRpcResponse) HasStream() bool

func (*WsJsonRpcResponse) Id 

func (w *WsJsonRpcResponse) Id() string

func (*WsJsonRpcResponse) ResponseCode 

func (w *WsJsonRpcResponse) ResponseCode() int

func (*WsJsonRpcResponse) ResponseResult 

func (w *WsJsonRpcResponse) ResponseResult() []byte

func (*WsJsonRpcResponse) ResponseResultString 

func (w *WsJsonRpcResponse) ResponseResultString() (string, error)

type WsResponse 

type WsResponse struct {
	Id         string
	SubId      string
	Message    []byte
	Type       RequestType
	Error      *ResponseError
	UpstreamId string
	// ParsedEvent is an optional, source-attached pre-parsed view of Message,
	ParsedEvent ParsedEvent
}

func ParseJsonRpcWsMessage 

func ParseJsonRpcWsMessage(body []byte) *WsResponse

func (*WsResponse) GetError 

func (w *WsResponse) GetError() *ResponseError

func (*WsResponse) GetMessage 

func (w *WsResponse) GetMessage() []byte

func (*WsResponse) GetParsedEvent 

func (w *WsResponse) GetParsedEvent() ParsedEvent

func (*WsResponse) GetUpstreamId 

func (w *WsResponse) GetUpstreamId() string

func (*WsResponse) IsEnd 

func (w *WsResponse) IsEnd() bool

IsEnd is false: a websocket subscription never ends cleanly on its own.

type ZeroBlock 

type ZeroBlock = Block

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.