Versions in this module Expand all Collapse all v0 v0.0.8 Aug 3, 2026 Changes in this version + const MetricQueryPhysicalConcurrent v0.0.7 Aug 3, 2026 Changes in this version + const DefaultCopyBatchMaxConcurrentCopies type CopyBatchConfig + MaxConcurrentCopies int v0.0.6 Aug 1, 2026 v0.0.5 Jul 31, 2026 Changes in this version + const AttributeCopyBatchFlushReason + const AttributeNodeName + const AttributeNodeRole + const AttributeRouteScope + const AttributeRouteShardCount + const AttributeShardName + const AttributeVirtualShard + const AttributeWrapperDelegated + const DefaultCopyBatchFlushTimeout + const MetricCopyBatchDuration + const MetricCopyBatchFlushes + const MetricCopyBatchRows + const MetricCopyBatchSubmissions + const MetricCopyQueueDuration + const MetricQueryLogicalDuration + const MetricQueryPhysicalDuration + const MetricQueryWrapperDuration + var ErrCopyBatchCountMismatch = errors.New("pgmesh: copy batch row count mismatch") + type CopyBatchConfig struct + BatchSize int + FlushTimeout time.Duration + func (c CopyBatchConfig) Validate() error + type CopyBatchExecutor func(context.Context, []T) (int64, error) + type CopyBatchFlushReason string + const CopyBatchFlushReasonExplicit + const CopyBatchFlushReasonImmediate + const CopyBatchFlushReasonSize + const CopyBatchFlushReasonTimeout + type CopyBatchObservation struct + Duration time.Duration + Err error + FlushReason CopyBatchFlushReason + QueueDuration time.Duration + Rows int + Submissions int + type CopyBatchObserver func(context.Context, CopyBatchObservation) + type CopyBatcher struct + func NewCopyBatcher[T any](config CopyBatchConfig, execute CopyBatchExecutor[T], ...) (*CopyBatcher[T], error) + func (b *CopyBatcher[T]) Flush(ctx context.Context) error + func (b *CopyBatcher[T]) FlushAsync() *Future[struct{}] + func (b *CopyBatcher[T]) Submit(ctx context.Context, rows []T) *Future[int64] + func (b *CopyBatcher[T]) SubmitImmediate(ctx context.Context, rows []T) *Future[int64] + type Future struct + func ResolvedFuture[T any](value T, err error) *Future[T] + func RunFuture[T any](fn func() (T, error)) *Future[T] + func (f *Future[T]) Await(ctx context.Context) (T, error) type Mesh + func (m *Mesh[R, W, SK]) CopyBatchObserver(storeName string, queryName string, route RouteMetadata) CopyBatchObserver + func (m *Mesh[R, W, SK]) StartQuerySpan(ctx context.Context, storeName string, queryName string, kind QueryKind, ...) (context.Context, *PhysicalQuerySpan) + type NodeRole string + const NodeRolePrimary + const NodeRoleReadReplica + const NodeRoleTransaction + type PhysicalQuerySpan struct + func (s *PhysicalQuerySpan) End(err error) type QuerySpan + func (s *QuerySpan) StartQuerySpan(ctx context.Context, route RouteMetadata, mode RouteMode) (context.Context, *PhysicalQuerySpan) + type Route struct + Node string + Role NodeRole + Shard string + Target T + VirtualShard uint64 + func (r Route[T]) Metadata() RouteMetadata + type RouteMetadata struct + HasVirtualShard bool + Node string + Role NodeRole + Shard string + VirtualShard uint64 + func (r RouteMetadata) WithoutVirtualShard() RouteMetadata type RouteMode + const RouteModeUnresolved + type RouteScope string + const RouteScopeFanout + const RouteScopeSingle + const RouteScopeUnresolved type Shard + func (s *Shard[R, W]) ReadRoute() Route[R] + func (s *Shard[R, W]) WriteRoute() Route[W] v0.0.4 Jul 29, 2026 v0.0.3 Jul 29, 2026 Changes in this version + const AttributeInternalStoreExecuted + const AttributeStoreName + const MetricStoreDuration type Mesh + func (m *Mesh[R, W, SK]) StartStoreSpan(ctx context.Context, storeName string, queryName string, kind QueryKind) (context.Context, *StoreSpan) + type StoreSpan struct + func (s *StoreSpan) End(err error) v0.0.2 Jul 28, 2026 v0.0.1 Jul 28, 2026 Changes in this version + const AttributeQueryKind + const AttributeQueryName + const AttributeReplicaSet + const AttributeRouteMode + const MetricQueryDuration + var ErrCrossShardTransaction = errors.New("pgmesh: transaction cannot span physical shards") + var ErrDuplicateReplicaSet = errors.New("pgmesh: duplicate replica set") + var ErrDuplicateVShard = errors.New("pgmesh: virtual shard is already linked") + var ErrEmptyDSN = errors.New("pgmesh: connection DSN must not be empty") + var ErrEmptyReplicaSetName = errors.New("pgmesh: replica set name must not be empty") + var ErrMirrorConfiguration = errors.New("pgmesh: inconsistent mirror configuration") + var ErrMissingVShard = errors.New("pgmesh: virtual shard is not linked") + var ErrNilReplicaSet = errors.New("pgmesh: replica set must not be nil") + var ErrNoNodeFactory = errors.New("pgmesh: node factory is required") + var ErrNoReplicaSets = errors.New("pgmesh: at least one replica set is required") + var ErrNoShardHasher = errors.New("pgmesh: shard hasher is required") + var ErrNoVShards = errors.New("pgmesh: at least one virtual shard is required") + var ErrUnknownReplicaSet = errors.New("pgmesh: unknown replica set") + var ErrVShardOutOfRange = errors.New("pgmesh: virtual shard is out of range") + func VShardRange(from, to uint64) []uint64 + type Builder struct + func NewBuilder[R any, W Mirrorable[W], SK any](numVShards uint64) *Builder[R, W, SK] + func (b *Builder[R, W, SK]) Build() (*Mesh[R, W, SK], error) + func (b *Builder[R, W, SK]) Link(vshard uint64, rs *ReplicaSet[R, W]) *Builder[R, W, SK] + func (b *Builder[R, W, SK]) WithHasher(hasher ShardHasher[SK]) *Builder[R, W, SK] + func (b *Builder[R, W, SK]) WithLogger(logger *slog.Logger) *Builder[R, W, SK] + func (b *Builder[R, W, SK]) WithMeterProvider(provider metric.MeterProvider) *Builder[R, W, SK] + func (b *Builder[R, W, SK]) WithTracerProvider(provider trace.TracerProvider) *Builder[R, W, SK] + type IntShardKey interface + type Mesh struct + func CreateMesh[R any, W Mirrorable[W], SK any](ctx context.Context, numVShards uint64, createNode NodeFactory[R, W], ...) (*Mesh[R, W, SK], error) + func (m *Mesh[R, W, SK]) AllShards() []*Shard[R, W] + func (m *Mesh[R, W, SK]) Shard(key SK) (*Shard[R, W], error) + func (m *Mesh[R, W, SK]) StartSpan(ctx context.Context, storeName string, queryName string, kind QueryKind) (context.Context, *QuerySpan) + type MeshOption func(*meshConfig) + func WithLogger(logger *slog.Logger) MeshOption + func WithMeterProvider(provider metric.MeterProvider) MeshOption + func WithReplicaSet(name, primaryDSN string, replicaDSNs ...string) MeshOption + func WithTracerProvider(provider trace.TracerProvider) MeshOption + func WithVShardMapping(mainReplicaSet string, vshards []uint64, mirrorReplicaSets ...string) MeshOption + type Mirrorable interface + WithMirrors func(...W) W + type Node struct + func NewNode[R any, W Mirrorable[W]](reader R, writer W) Node[R, W] + func (n Node[R, W]) Reader() R + func (n Node[R, W]) Writer() W + type NodeFactory func(context.Context, string) (Node[R, W], error) + type QueryKind string + const QueryKindRead + const QueryKindWrite + type QuerySpan struct + func (s *QuerySpan) End(err error) + func (s *QuerySpan) SetMultiRoute(mode RouteMode) + func (s *QuerySpan) SetRoute(vshard uint64, replicaSet string, mode RouteMode) + type ReplicaSet struct + func NewReplicaSet[R any, W Mirrorable[W]](name string, primary Node[R, W], replicas []Node[R, W]) *ReplicaSet[R, W] + func (s *ReplicaSet[R, W]) Name() string + func (s *ReplicaSet[R, W]) Read() R + func (s *ReplicaSet[R, W]) WithWriteMirrors(writes ...W) *ReplicaSet[R, W] + func (s *ReplicaSet[R, W]) Write() W + func (s *ReplicaSet[R, W]) WriteMirrorCount() int + type RouteMode string + const RouteModePrimary + const RouteModeRead + const RouteModeTransaction + type Shard struct + func (s *Shard[R, W]) VShardIndex() uint64 + type ShardHasher interface + Hash func(SK) uint64 + func ConstantShardHashFor[SK any](vshard uint64) ShardHasher[SK] + func ModularShardHashFor[SK IntShardKey](numVShards uint64) ShardHasher[SK]
Go to main page