network

package
v0.9.0 Latest Latest
Warning

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

 Go to latest Published: Sep 4, 2026 License: MIT Imports: 27 Imported by: 0

Documentation

Index

Constants

View Source 
const MaxRequestBodyBytes = 1 << 20

MaxRequestBodyBytes is the HTTP adapter's JSON decoding limit. Replicated mutations have the smaller quepaxa.MaxReplicatedValueBytes consensus limit.

Variables

View Source 
var (
	ErrNotReady              = errors.New("node is not ready")
	ErrRequestConflict       = errors.New("request ID conflict")
	ErrInvalidRequest        = errors.New("invalid request")
	ErrOverloaded            = errors.New("mutation queue overloaded")
	ErrDurabilityUnavailable = errors.New("object-store durability unavailable")
	ErrCommitUnknown         = errors.New("commit outcome unknown")
)

Functions

func ValidateExecuteRequest added in v0.8.1 

func ValidateExecuteRequest(req ExecuteRequest) error

ValidateExecuteRequest applies the same mutation contract and encoded-size check as Execute without submitting the command.

Types

type CommitUnknownError 

type CommitUnknownError struct {
	Slot             quepaxa.Slot
	RequestID        string
	RetryThroughSlot uint64
	Cause            error
}

CommitUnknownError means a mutation may commit despite the failed call. Retrying the same request ID resolves the outcome without duplicating it.

func (*CommitUnknownError) Error 

func (e *CommitUnknownError) Error() string

func (*CommitUnknownError) Unwrap 

func (e *CommitUnknownError) Unwrap() []error

type DecisionsResponse 

type DecisionsResponse struct {
	ClusterID  types.ClusterID        `json:"cluster_id"`
	ProposerID quepaxa.NodeID         `json:"proposer_id"`
	ConfigID   uint                   `json:"config_id"`
	Tip        quepaxa.Slot           `json:"tip"`
	Decisions  []quepaxa.DecidedValue `json:"decisions"`
}

type ExecuteRequest 

type ExecuteRequest struct {
	RequestID string `json:"request_id"`
	SQL       string `json:"sql,omitempty"`
	Args      []any  `json:"args,omitempty"`
	// WantRows is unsupported for replicated mutations; use Query after Execute.
	WantRows   bool                 `json:"want_rows,omitempty"`
	Statements []types.SQLStatement `json:"statements,omitempty"`
}

ExecuteRequest is the request body for execute.

type ExecuteResponse 

type ExecuteResponse struct {
	types.MutationReceipt
}

ExecuteResponse contains the bounded aggregate receipt retained for retries. Replicated statement rows are not returned; use Query after Execute.

type GraphExecuteResponse 

type GraphExecuteResponse struct {
	types.MutationReceipt
}

type GraphQueryRequest 

type GraphQueryRequest struct {
	Cypher      string         `json:"cypher"`
	Args        map[string]any `json:"args,omitempty"`
	Consistency string         `json:"consistency,omitempty"`
}

type GraphStreamOffsetRequest 

type GraphStreamOffsetRequest struct {
	RequestID   string `json:"request_id,omitempty"`
	Stream      string `json:"stream"`
	Consumer    string `json:"consumer"`
	Sequence    uint64 `json:"sequence,omitempty"`
	Consistency string `json:"consistency,omitempty"`
}

type GraphStreamOffsetResponse 

type GraphStreamOffsetResponse struct {
	Sequence     uint64 `json:"sequence"`
	Found        bool   `json:"found"`
	AppliedSlot  uint64 `json:"applied_slot"`
	ConsensusTip uint64 `json:"consensus_tip"`
}

type GraphStreamReadRequest 

type GraphStreamReadRequest struct {
	Stream        string `json:"stream,omitempty"`
	AfterSequence uint64 `json:"after_sequence,omitempty"`
	Limit         uint   `json:"limit,omitempty"`
	WaitMS        uint32 `json:"wait_ms,omitempty"`
	Consistency   string `json:"consistency,omitempty"`
}

type GraphStreamReadResponse 

type GraphStreamReadResponse struct {
	Records      []types.GraphStreamRecord `json:"records"`
	AppliedSlot  uint64                    `json:"applied_slot"`
	ConsensusTip uint64                    `json:"consensus_tip"`
}

type GraphStreamTrimRequest 

type GraphStreamTrimRequest struct {
	RequestID       string `json:"request_id"`
	Stream          string `json:"stream"`
	ThroughSequence uint64 `json:"through_sequence"`
}

type KVGetRequest 

type KVGetRequest struct{ Key, Consistency string }

type KVGetResponse 

type KVGetResponse struct {
	Found        bool   `json:"found"`
	Value        []byte `json:"value,omitempty"`
	AppliedSlot  uint64 `json:"applied_slot"`
	ConsensusTip uint64 `json:"consensus_tip"`
}

type KVMutationRequest 

type KVMutationRequest struct {
	RequestID      string `json:"request_id"`
	Key            string `json:"key"`
	Value          []byte `json:"value,omitempty"`
	Expected       []byte `json:"expected,omitempty"`
	ExpectedExists bool   `json:"expected_exists,omitempty"`
	TTLMS          int64  `json:"ttl_ms,omitempty"`
}

type KVMutationResponse 

type KVMutationResponse struct {
	types.MutationReceipt
}

type PeerServer 

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

PeerServer owns the private QUIC listener. Public HTTP remains a separate adapter.

func StartPeerServer 

func StartPeerServer(ctx context.Context, addr string, server *Server, members []quepaxa.Member, token string) (*PeerServer, error)

func (*PeerServer) Addr 

func (s *PeerServer) Addr() string

func (*PeerServer) Close 

func (s *PeerServer) Close() error

type QueryRequest 

type QueryRequest struct {
	SQL         string `json:"sql"`
	Args        []any  `json:"args,omitempty"`
	Consistency string `json:"consistency,omitempty"`
}

QueryRequest is the request body for query.

type QueryResponse 

type QueryResponse struct {
	Columns      []string        `json:"columns"`
	Rows         [][]interface{} `json:"rows"`
	AppliedSlot  uint64          `json:"applied_slot"`
	ConsensusTip uint64          `json:"consensus_tip"`
}

QueryResponse is the response body for query.

type RequestStatusRequest 

type RequestStatusRequest struct {
	Kind      string `json:"kind"`
	RequestID string `json:"request_id"`
}

type RequestStatusResponse 

type RequestStatusResponse struct {
	State   string                 `json:"state"`
	Tip     uint64                 `json:"tip"`
	Receipt *types.MutationReceipt `json:"receipt,omitempty"`
}

type Server 

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

Server is the HTTP server for client API.

func NewServer 

func NewServer(core *quepaxa.Core, material *materializer.Materializer, cluster types.ClusterID, writable bool, transport *Transport, members []quepaxa.Member, hedgeDelay time.Duration, ready ...func() bool) *Server

NewServer creates a new HTTP server.

func (*Server) Close 

func (s *Server) Close()

Close stops background request batching.

func (*Server) Execute 

func (s *Server) Execute(ctx context.Context, req ExecuteRequest) (ExecuteResponse, error)

Execute applies one SQL statement or an atomic statements transaction.

func (*Server) GraphChanges 

func (s *Server) GraphChanges(ctx context.Context, request GraphStreamReadRequest) (GraphStreamReadResponse, error)

func (*Server) GraphExecute 

func (s *Server) GraphExecute(ctx context.Context, command types.GraphCommand) (GraphExecuteResponse, error)

func (*Server) GraphQuery 

func (s *Server) GraphQuery(ctx context.Context, request GraphQueryRequest) (types.GraphCommandResult, error)

func (*Server) GraphStreamOffset 

func (s *Server) GraphStreamOffset(ctx context.Context, request GraphStreamOffsetRequest) (GraphStreamOffsetResponse, error)

func (*Server) GraphStreamRead 

func (s *Server) GraphStreamRead(ctx context.Context, request GraphStreamReadRequest) (GraphStreamReadResponse, error)

func (*Server) KVCAS 

func (s *Server) KVCAS(ctx context.Context, req KVMutationRequest) (KVMutationResponse, error)

func (*Server) KVDelete 

func (s *Server) KVDelete(ctx context.Context, req KVMutationRequest) (KVMutationResponse, error)

func (*Server) KVGet 

func (s *Server) KVGet(ctx context.Context, req KVGetRequest) (KVGetResponse, error)

func (*Server) KVMutate 

func (s *Server) KVMutate(ctx context.Context, operation string, req KVMutationRequest) (KVMutationResponse, error)

func (*Server) KVPut 

func (s *Server) KVPut(ctx context.Context, req KVMutationRequest) (KVMutationResponse, error)

func (*Server) NotificationDrops 

func (s *Server) NotificationDrops() uint64

func (*Server) NotifyPublish 

func (s *Server) NotifyPublish(ctx context.Context, req types.NotifyCommand) (types.MutationReceipt, error)

func (*Server) NotifySubscribe 

func (s *Server) NotifySubscribe(topic string) (<-chan []byte, func(), error)

func (*Server) ProposeControl 

func (s *Server) ProposeControl(ctx context.Context, value []byte) (quepaxa.Slot, error)

ProposeControl commits an internal read barrier through the normal bounded proposal lifecycle.

func (*Server) Query 

func (s *Server) Query(ctx context.Context, req QueryRequest) (QueryResponse, error)

Query reads SQL locally or after a linearizable consensus barrier.

func (*Server) Quiesce 

func (s *Server) Quiesce(ctx context.Context) (func(), error)

Quiesce drains proposals and excludes decision application while a certified checkpoint replaces local consensus and materialized state.

func (*Server) RequestStatus 

func (s *Server) RequestStatus(ctx context.Context, req RequestStatusRequest) (RequestStatusResponse, error)

func (*Server) ServeHTTP 

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP implements http.Handler.

func (*Server) SetCheckpointPrepare 

func (s *Server) SetCheckpointPrepare(prepare func(context.Context, quepaxa.NodeID, quepaxa.CheckpointSeal) error)

func (*Server) SetCompactedHandler 

func (s *Server) SetCompactedHandler(handler func())

SetCompactedHandler installs the recovery trigger used when a peer has compacted history this node still needs.

func (*Server) SetDurabilityBarrier 

func (s *Server) SetDurabilityBarrier(barrier func(context.Context, quepaxa.Slot) error)

SetDurabilityBarrier installs the mutation ACK barrier before the server is exposed.

func (*Server) SetGraphStreamOffset 

func (s *Server) SetGraphStreamOffset(ctx context.Context, request GraphStreamOffsetRequest) error

func (*Server) SetObjectStoreStats 

func (s *Server) SetObjectStoreStats(stats func() (map[string]uint64, bool))

func (*Server) TrimGraphStream 

func (s *Server) TrimGraphStream(ctx context.Context, request GraphStreamTrimRequest) error

type Transport 

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

Transport sends private peer RPCs over persistent raw QUIC connections.

func NewTransport 

func NewTransport(clusterID types.ClusterID, localID quepaxa.NodeID, config *quepaxa.Cluster, token string) *Transport

func (*Transport) Close 

func (t *Transport) Close() error

func (*Transport) FetchDecisions 

func (t *Transport) FetchDecisions(ctx context.Context, source quepaxa.NodeID, from quepaxa.Slot, limit int) (DecisionsResponse, error)

func (*Transport) FetchValue 

func (t *Transport) FetchValue(ctx context.Context, from quepaxa.NodeID, hash quepaxa.ValueHash) ([]byte, error)

func (*Transport) PrepareCheckpoint 

func (t *Transport) PrepareCheckpoint(ctx context.Context, seal quepaxa.CheckpointSeal) error

PrepareCheckpoint waits for a durable verified quorum before the small seal value enters normal consensus.

func (*Transport) Propose 

func (t *Transport) Propose(ctx context.Context, to quepaxa.NodeID, value []byte) (quepaxa.DecidedValue, error)

func (*Transport) ReadTip 

func (t *Transport) ReadTip(ctx context.Context, to quepaxa.NodeID) (quepaxa.Slot, error)

func (*Transport) SendDecision 

func (t *Transport) SendDecision(ctx context.Context, decision quepaxa.Decision) error

func (*Transport) SendRecord 

func (t *Transport) SendRecord(ctx context.Context, to quepaxa.NodeID, request quepaxa.RecordRequest) (quepaxa.Summary, error)

func (*Transport) StageValue 

func (t *Transport) StageValue(ctx context.Context, to quepaxa.NodeID, hash quepaxa.ValueHash, value []byte) error

Source Files

View all Source files

Directories

Path Synopsis

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.