pgmesh

package module
v0.0.8 Latest Latest
Warning

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

 Go to latest Published: Aug 3, 2026 License: MIT Imports: 15 Imported by: 0

README

pgmesh

Test Lint Integration

Type-safe PostgreSQL query routing for sqlc and pgx/v5.

Why

sqlc generates type-safe queries, but it does not express where each query may run. As an application adds read replicas or shards, routing logic can spread through business code and make it easy to send a write to the wrong database.

pgmesh keeps that policy in generated Go types and one validated runtime topology.

What

pgmesh is a sqlc process plugin plus a small Go runtime. Together they provide:

  • separate read, write, and primary-capable query APIs;
  • mandatory query groups for keeping large generated stores navigable;
  • replica reads with explicit primary reads when consistency requires them;
  • logical-key routing through virtual shards to physical databases;
  • explicit scatter queries and grouped list lookups;
  • synchronous and asynchronous per-shard COPY FROM, with configurable micro-batching and explicit flush barriers;
  • query-group factories for cache-aside and other application-specific wrappers;
  • shard-pinned transactions;
  • synchronous write mirrors for staged shard expansion; and
  • OpenTelemetry instrumentation and structured debug logging.

It is not a database proxy, connection pool, replication system, data migration tool, or distributed transaction coordinator. Your application keeps control of those concerns.

How

Install the runtime and generator:

go get github.com/sundayfun/pgmesh
go install github.com/sundayfun/pgmesh/cmd/sqlc-gen-store@latest

Classify and group each sqlc query. Add a shard route only when the query should be routed automatically:

-- name: GetAccount :one
-- kind: read
-- shard: tenantKey(tenant_id)
-- store: Accounts
SELECT * FROM accounts WHERE tenant_id = $1 AND id = $2;

-- name: UpsertAccount :one
-- kind: write
-- shard: tenantKey(tenant_id)
-- store: Accounts
INSERT INTO accounts (id, tenant_id, display_name) VALUES ($1, $2, $3)
ON CONFLICT (id) DO UPDATE SET display_name = EXCLUDED.display_name
RETURNING *;

Register the process plugin in sqlc.yaml, then generate both sqlc's queries and pgmesh's wrappers:

sqlc generate

Construct the generated Store with a singleton topology:

queries, err := db.NewStore(ctx, db.Singleton(pool))
account, err := queries.Accounts().GetAccount(ctx, &db.GetAccountT{
    TenantKey: db.TenantKey{TenantID: tenantID},
    ID:        accountID,
})

When the deployment grows, replace Singleton(...) with a Sharded(...) topology. queries remains the same db.Store interface, so business code does not depend on whether pgmesh uses one database, replicas, mirrors, or shards.

Follow the quickstart for a complete working setup, or explore the topology and focused feature examples, including replicas, sharding, write mirrors, transactions, cache-aside, asynchronous COPY batching, and multi-shard queries.

Documentation

License

MIT

Documentation

Overview

Package pgmesh provides type-safe replica routing and virtual sharding for query wrappers generated by the pgmesh process plugin.

Mesh values are immutable after construction and may be shared by concurrent callers when their configured nodes are concurrency-safe. Applications own database pools, loggers, and OpenTelemetry providers; pgmesh does not close or shut them down.

Index

Examples

Constants

View Source 
const (
	// AttributeStoreName identifies the generated store query group.
	AttributeStoreName = "pgmesh.store.name"
	// AttributeQueryName identifies the generated query method.
	AttributeQueryName = "pgmesh.query.name"
	// AttributeQueryKind identifies whether a routed query is a read or write.
	AttributeQueryKind = "pgmesh.query.kind"
	// AttributeShardName identifies the selected physical shard (replica set).
	AttributeShardName = "pgmesh.shard.name"
	// AttributeVirtualShard identifies the selected virtual shard on spans and
	// logs. It is deliberately excluded from metrics to bound cardinality.
	AttributeVirtualShard = "pgmesh.shard.virtual"
	// AttributeNodeName identifies a stable node within a physical shard.
	AttributeNodeName = "pgmesh.node.name"
	// AttributeNodeRole identifies whether the node is a primary or read replica.
	AttributeNodeRole = "pgmesh.node.role"
	// AttributeRouteMode identifies the database path selected for a query.
	AttributeRouteMode = "pgmesh.route.mode"
	// AttributeRouteScope identifies single-shard versus fan-out operations.
	AttributeRouteScope = "pgmesh.route.scope"
	// AttributeRouteShardCount reports the number of physical shard executions
	// on spans and logs. It is excluded from metrics to bound cardinality.
	AttributeRouteShardCount = "pgmesh.route.shard_count"
	// AttributeWrapperDelegated reports whether an application wrapper
	// delegated to the generated logical query implementation.
	AttributeWrapperDelegated = "pgmesh.wrapper.delegated"
	// AttributeCopyBatchFlushReason identifies why a physical COPY batch was
	// made ready for execution.
	AttributeCopyBatchFlushReason = "pgmesh.copy.batch.flush_reason"
)

OpenTelemetry attribute keys recorded on store and routed query telemetry.

View Source 
const DefaultCopyBatchFlushTimeout = time.Millisecond

DefaultCopyBatchFlushTimeout is used when CopyBatchConfig.FlushTimeout is zero.

View Source 
const DefaultCopyBatchMaxConcurrentCopies = 32

DefaultCopyBatchMaxConcurrentCopies is used when CopyBatchConfig.MaxConcurrentCopies is zero.

View Source 
const MetricCopyBatchDuration = "pgmesh.copy.batch.duration"

MetricCopyBatchDuration is the OpenTelemetry histogram of physical COPY execution durations in seconds.

View Source 
const MetricCopyBatchFlushes = "pgmesh.copy.batch.flushes"

MetricCopyBatchFlushes counts physical COPY operations by flush reason.

View Source 
const MetricCopyBatchRows = "pgmesh.copy.batch.rows"

MetricCopyBatchRows is the OpenTelemetry histogram of attempted rows per physical COPY operation.

View Source 
const MetricCopyBatchSubmissions = "pgmesh.copy.batch.submissions"

MetricCopyBatchSubmissions is the OpenTelemetry histogram of logical submission fragments represented in each physical COPY operation.

View Source 
const MetricCopyQueueDuration = "pgmesh.copy.queue.duration"

MetricCopyQueueDuration is the OpenTelemetry histogram of time from the oldest row's admission until physical COPY execution begins, in seconds.

View Source 
const MetricQueryLogicalDuration = "pgmesh.query.logical.duration"

MetricQueryLogicalDuration is the OpenTelemetry histogram of logical generated-query durations in seconds. Fan-out work contributes one data point.

View Source 
const MetricQueryPhysicalConcurrent = "pgmesh.query.physical.concurrent"

MetricQueryPhysicalConcurrent is the OpenTelemetry up-down counter of physical database queries currently executing. It is grouped by the same bounded query and target attributes as MetricQueryPhysicalDuration.

View Source 
const MetricQueryPhysicalDuration = "pgmesh.query.physical.duration"

MetricQueryPhysicalDuration is the OpenTelemetry histogram of physical database-query durations in seconds. Its count reports per-node query throughput. The configured MeterProvider owns exporting and shutdown; pgmesh never shuts it down.

View Source 
const MetricQueryWrapperDuration = "pgmesh.query.wrapper.duration"

MetricQueryWrapperDuration is the OpenTelemetry histogram of optional application wrapper durations in seconds. Its count reports completed wrapper throughput.

Variables

View Source 
var (
	// ErrNoReplicaSets indicates that a topology contains no replica sets.
	ErrNoReplicaSets = errors.New("pgmesh: at least one replica set is required")
	// ErrEmptyReplicaSetName indicates that a replica set has no name.
	ErrEmptyReplicaSetName = errors.New("pgmesh: replica set name must not be empty")
	// ErrDuplicateReplicaSet indicates that a topology reuses a replica set name.
	ErrDuplicateReplicaSet = errors.New("pgmesh: duplicate replica set")
	// ErrEmptyDSN indicates that a database connection has no DSN.
	ErrEmptyDSN = errors.New("pgmesh: connection DSN must not be empty")
	// ErrNoVShards indicates that a topology contains no virtual shards.
	ErrNoVShards = errors.New("pgmesh: at least one virtual shard is required")
	// ErrDuplicateVShard indicates that a virtual shard has already been linked.
	ErrDuplicateVShard = errors.New("pgmesh: virtual shard is already linked")
	// ErrMissingVShard indicates that a virtual shard has not been linked.
	ErrMissingVShard = errors.New("pgmesh: virtual shard is not linked")
	// ErrVShardOutOfRange indicates that a virtual shard index is outside the topology.
	ErrVShardOutOfRange = errors.New("pgmesh: virtual shard is out of range")
	// ErrNoShardHasher indicates that no shard-key hasher was configured.
	ErrNoShardHasher = errors.New("pgmesh: shard hasher is required")
	// ErrNoNodeFactory indicates that no database node factory was configured.
	ErrNoNodeFactory = errors.New("pgmesh: node factory is required")
	// ErrUnknownReplicaSet indicates that a shard mapping names an undefined replica set.
	ErrUnknownReplicaSet = errors.New("pgmesh: unknown replica set")
	// ErrNilReplicaSet indicates that a builder was given a nil replica set.
	ErrNilReplicaSet = errors.New("pgmesh: replica set must not be nil")
	// ErrMirrorConfiguration indicates that write-mirror mappings are inconsistent.
	ErrMirrorConfiguration = errors.New("pgmesh: inconsistent mirror configuration")
	// ErrCrossShardTransaction indicates that one transaction was supplied for
	// an operation targeting more than one physical shard.
	ErrCrossShardTransaction = errors.New("pgmesh: transaction cannot span physical shards")
)
View Source 
var ErrCopyBatchCountMismatch = errors.New("pgmesh: copy batch row count mismatch")

ErrCopyBatchCountMismatch reports a successful physical COPY whose returned row count does not match the number of rows supplied to it.

Functions

func VShardRange 

func VShardRange(from, to uint64) []uint64

VShardRange returns the half-open virtual shard range [from, to).

Types

type Builder 

type Builder[R any, W Mirrorable[W], SK any] struct {
	// contains filtered or unexported fields
}

Builder incrementally assembles and validates an immutable Mesh topology. A Builder is intended for single-goroutine setup; the Mesh returned by Build can be shared by concurrent callers when its configured nodes can be shared.

func NewBuilder 

func NewBuilder[R any, W Mirrorable[W], SK any](numVShards uint64) *Builder[R, W, SK]

NewBuilder creates a builder with numVShards unlinked virtual shards.

Example 
package main

import (
	"fmt"

	"github.com/sundayfun/pgmesh"
)

type exampleReader struct {
	node string
}

type exampleWriter struct {
	node    string
	mirrors []*exampleWriter
}

func (q *exampleWriter) WithMirrors(mirrors ...*exampleWriter) *exampleWriter {
	return &exampleWriter{
		node:    q.node,
		mirrors: append(append([]*exampleWriter(nil), q.mirrors...), mirrors...),
	}
}

func (q *exampleWriter) Put(value string) []string {
	writes := []string{q.node + ":" + value}
	for _, mirror := range q.mirrors {
		writes = append(writes, mirror.node+":"+value)
	}
	return writes
}

func exampleNode(name string) pgmesh.Node[*exampleReader, *exampleWriter] {
	return pgmesh.NewNode(
		&exampleReader{node: name},
		&exampleWriter{node: name, mirrors: nil},
	)
}

func main() {
	shard0 := pgmesh.NewReplicaSet(
		"shard-0",
		exampleNode("shard0-primary"),
		[]pgmesh.Node[*exampleReader, *exampleWriter]{
			exampleNode("shard0-replica0"),
			exampleNode("shard0-replica1"),
		},
	)
	shard1 := pgmesh.NewReplicaSet("shard-1", exampleNode("shard1-primary"), nil)

	mesh, err := pgmesh.NewBuilder[*exampleReader, *exampleWriter, uint64](2).
		WithHasher(pgmesh.ModularShardHashFor[uint64](2)).
		Link(0, shard0).
		Link(1, shard1).
		Build()
	if err != nil {
		panic(err)
	}

	routed, err := mesh.Shard(2)
	if err != nil {
		panic(err)
	}
	fmt.Println(routed.Name(), routed.VShardIndex())
	fmt.Println(routed.Read().node)
	fmt.Println(routed.Read().node)
	fmt.Println(routed.Write().Put("message"))

	fallback, err := mesh.Shard(3)
	if err != nil {
		panic(err)
	}
	fmt.Println(fallback.Read().node)

	for _, shard := range mesh.AllShards() {
		fmt.Println(shard.Name())
	}

}

Output:
shard-0 0
shard0-replica0
shard0-replica1
[shard0-primary:message]
shard1-primary
shard-0
shard-1

func (*Builder[R, W, SK]) Build 

func (b *Builder[R, W, SK]) Build() (*Mesh[R, W, SK], error)

Build validates the topology and returns an immutable mesh. It retains the configured node and telemetry providers; callers remain responsible for shutting down database pools and OpenTelemetry SDK providers. 

func (b *Builder[R, W, SK]) Link(vshard uint64, rs *ReplicaSet[R, W]) *Builder[R, W, SK]

Link records validation failures and returns the builder so topology setup remains fluent without panics. Build returns the first recorded error.

func (*Builder[R, W, SK]) WithHasher 

func (b *Builder[R, W, SK]) WithHasher(hasher ShardHasher[SK]) *Builder[R, W, SK]

WithHasher configures the mapping from shard keys to virtual shard indexes.

func (*Builder[R, W, SK]) WithLogger 

func (b *Builder[R, W, SK]) WithLogger(logger *slog.Logger) *Builder[R, W, SK]

WithLogger configures optional structured logging for routed queries. Completed queries are logged at Debug level. A nil logger disables logging.

func (*Builder[R, W, SK]) WithMeterProvider 

func (b *Builder[R, W, SK]) WithMeterProvider(provider metric.MeterProvider) *Builder[R, W, SK]

WithMeterProvider configures the provider used for routed query metrics. A nil provider uses the global OpenTelemetry meter provider.

func (*Builder[R, W, SK]) WithTracerProvider 

func (b *Builder[R, W, SK]) WithTracerProvider(provider trace.TracerProvider) *Builder[R, W, SK]

WithTracerProvider configures the provider used for routed query spans. A nil provider uses the global OpenTelemetry tracer provider.

type CopyBatchConfig added in v0.0.5 

type CopyBatchConfig struct {
	// BatchSize is the maximum number of rows in one physical COPY. Zero leaves
	// timed and backlog-merged batches unbounded by row count.
	BatchSize int
	// FlushTimeout is measured from the first row in a partial batch. Zero uses
	// DefaultCopyBatchFlushTimeout.
	FlushTimeout time.Duration
	// MaxConcurrentCopies is the maximum number of physical COPY executions in
	// flight for one database target. Zero uses
	// DefaultCopyBatchMaxConcurrentCopies.
	MaxConcurrentCopies int
}

CopyBatchConfig controls how rows submitted by concurrent callers are coalesced into physical COPY operations.

func (CopyBatchConfig) Validate added in v0.0.5 

func (c CopyBatchConfig) Validate() error

Validate validates the configuration without creating a batcher.

type CopyBatchExecutor added in v0.0.5 

type CopyBatchExecutor[T any] func(context.Context, []T) (int64, error)

CopyBatchExecutor performs one physical COPY for rows on a single target.

type CopyBatchFlushReason added in v0.0.5 

type CopyBatchFlushReason string

CopyBatchFlushReason identifies the boundary that made a physical COPY batch ready for execution. 

const (
	CopyBatchFlushReasonSize      CopyBatchFlushReason = "size"
	CopyBatchFlushReasonTimeout   CopyBatchFlushReason = "timeout"
	CopyBatchFlushReasonExplicit  CopyBatchFlushReason = "explicit"
	CopyBatchFlushReasonImmediate CopyBatchFlushReason = "immediate"
)

Physical COPY batch flush reasons.

type CopyBatchObservation added in v0.0.5 

type CopyBatchObservation struct {
	Rows          int
	Submissions   int
	FlushReason   CopyBatchFlushReason
	QueueDuration time.Duration
	Duration      time.Duration
	Err           error
}

CopyBatchObservation describes one completed physical COPY batch. Rows is the attempted physical batch size. Submissions is the number of logical submission fragments represented in that batch.

type CopyBatchObserver added in v0.0.5 

type CopyBatchObserver func(context.Context, CopyBatchObservation)

CopyBatchObserver receives completed physical COPY batch observations. A batcher may invoke an observer concurrently.

type CopyBatcher added in v0.0.5 

type CopyBatcher[T any] struct {
	// contains filtered or unexported fields
}

CopyBatcher coalesces rows for one generated COPY query and one physical database target. It executes up to CopyBatchConfig.MaxConcurrentCopies physical COPY operations at a time.

func NewCopyBatcher added in v0.0.5 

func NewCopyBatcher[T any](
	config CopyBatchConfig,
	execute CopyBatchExecutor[T],
	observers ...CopyBatchObserver,
) (*CopyBatcher[T], error)

NewCopyBatcher validates config and creates a batcher for one physical COPY target. execute must not be nil.

func (*CopyBatcher[T]) Flush added in v0.0.5 

func (b *CopyBatcher[T]) Flush(ctx context.Context) error

Flush forces the current partial batch to execute and waits for submissions that were outstanding at the flush barrier. Submissions accepted later are not included.

func (*CopyBatcher[T]) FlushAsync added in v0.0.5 

func (b *CopyBatcher[T]) FlushAsync() *Future[struct{}]

FlushAsync forces the current partial batch to execute and returns a Future for submissions that were outstanding at the flush barrier. Submissions accepted later are not included.

func (*CopyBatcher[T]) Submit added in v0.0.5 

func (b *CopyBatcher[T]) Submit(ctx context.Context, rows []T) *Future[int64]

Submit accepts rows for asynchronous COPY. Once Submit returns a pending Future, canceling ctx no longer cancels the accepted write. Callers must not mutate rows or referenced row data until the Future resolves.

func (*CopyBatcher[T]) SubmitImmediate added in v0.0.5 

func (b *CopyBatcher[T]) SubmitImmediate(ctx context.Context, rows []T) *Future[int64]

SubmitImmediate accepts rows as one physical COPY without coalescing them with other submissions. It shares the batcher's execution capacity and is included in Flush barriers, but concurrent batches may complete out of order. Callers must not mutate rows or referenced row data until the Future resolves.

type Future added in v0.0.5 

type Future[T any] struct {
	// contains filtered or unexported fields
}

Future is a repeatable, concurrency-safe handle for an asynchronous result. Canceling an Await call only stops that wait; it does not cancel the work.

func ResolvedFuture added in v0.0.5 

func ResolvedFuture[T any](value T, err error) *Future[T]

ResolvedFuture returns a Future that has already completed.

func RunFuture added in v0.0.5 

func RunFuture[T any](fn func() (T, error)) *Future[T]

RunFuture starts fn asynchronously and resolves the returned Future with its result. The function owns any cancellation semantics for the work itself.

func (*Future[T]) Await added in v0.0.5 

func (f *Future[T]) Await(ctx context.Context) (T, error)

Await waits for the Future or for ctx to be canceled. A later Await may still retrieve the completed operation after an earlier wait was canceled.

type IntShardKey 

type IntShardKey interface {
	~int | ~int8 | ~int16 | ~int32 | ~int64 |
		~uint | ~uint8 | ~uint16 | ~uint32 | ~uint64
}

IntShardKey is the set of integer types supported by ModularShardHashFor.

type Mesh 

type Mesh[R any, W Mirrorable[W], SK any] struct {
	// contains filtered or unexported fields
}

Mesh routes logical shard keys through virtual shards to physical replica sets. Its topology is immutable after construction and safe for concurrent use.

func CreateMesh 

func CreateMesh[R any, W Mirrorable[W], SK any](
	ctx context.Context,
	numVShards uint64,
	createNode NodeFactory[R, W],
	shardHasher ShardHasher[SK],
	options ...MeshOption,
) (*Mesh[R, W, SK], error)

CreateMesh validates its configuration, opens its database nodes, and builds an immutable mesh. It calls createNode once for each primary and replica, in option order, and stops at the first error. Successfully created nodes are not closed on a later error and remain caller-owned.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/sundayfun/pgmesh"
)

type exampleReader struct {
	node string
}

type exampleWriter struct {
	node    string
	mirrors []*exampleWriter
}

func (q *exampleWriter) WithMirrors(mirrors ...*exampleWriter) *exampleWriter {
	return &exampleWriter{
		node:    q.node,
		mirrors: append(append([]*exampleWriter(nil), q.mirrors...), mirrors...),
	}
}

func (q *exampleWriter) Put(value string) []string {
	writes := []string{q.node + ":" + value}
	for _, mirror := range q.mirrors {
		writes = append(writes, mirror.node+":"+value)
	}
	return writes
}

func exampleNode(name string) pgmesh.Node[*exampleReader, *exampleWriter] {
	return pgmesh.NewNode(
		&exampleReader{node: name},
		&exampleWriter{node: name, mirrors: nil},
	)
}

func main() {
	mesh, err := pgmesh.CreateMesh(
		context.Background(),
		4,
		func(_ context.Context, dsn string) (
			pgmesh.Node[*exampleReader, *exampleWriter],
			error,
		) {
			return exampleNode(dsn), nil
		},
		pgmesh.ModularShardHashFor[uint64](4),
		pgmesh.WithReplicaSet("east", "east-primary", "east-replica"),
		pgmesh.WithReplicaSet("west", "west-primary"),
		pgmesh.WithReplicaSet("archive", "archive-primary"),
		pgmesh.WithVShardMapping("east", []uint64{0, 2}, "archive"),
		pgmesh.WithVShardMapping("west", []uint64{1, 3}),
	)
	if err != nil {
		panic(err)
	}

	routed, err := mesh.Shard(6)
	if err != nil {
		panic(err)
	}
	fmt.Println(routed.Name(), routed.VShardIndex())
	fmt.Println(routed.Read().node)
	fmt.Println(routed.Write().Put("event"))

}

Output:
east 2
east-replica
[east-primary:event archive-primary:event]

func (*Mesh[R, W, SK]) AllShards 

func (m *Mesh[R, W, SK]) AllShards() []*Shard[R, W]

AllShards returns one entry per physical replica set in first-vshard order.

func (*Mesh[R, W, SK]) CopyBatchObserver added in v0.0.5 

func (m *Mesh[R, W, SK]) CopyBatchObserver(
	storeName string,
	queryName string,
	route RouteMetadata,
) CopyBatchObserver

CopyBatchObserver returns an observer used by generated stores to record one metric point for every physical COPY operation on a replica set.

func (*Mesh[R, W, SK]) Shard 

func (m *Mesh[R, W, SK]) Shard(key SK) (*Shard[R, W], error)

Shard resolves key to its virtual shard and physical replica set.

func (*Mesh[R, W, SK]) StartQuerySpan added in v0.0.5 

func (m *Mesh[R, W, SK]) StartQuerySpan(
	ctx context.Context,
	storeName string,
	queryName string,
	kind QueryKind,
	route RouteMetadata,
	mode RouteMode,
) (context.Context, *PhysicalQuerySpan)

StartQuerySpan starts a physical database-query span without requiring a logical QuerySpan. It is used for asynchronous COPY batches.

func (*Mesh[R, W, SK]) StartSpan 

func (m *Mesh[R, W, SK]) StartSpan(
	ctx context.Context,
	storeName string,
	queryName string,
	kind QueryKind,
) (context.Context, *QuerySpan)

StartSpan starts telemetry for one logical generated operation. Physical executions are recorded as child spans through StartQuerySpan.

func (*Mesh[R, W, SK]) StartStoreSpan added in v0.0.3 

func (m *Mesh[R, W, SK]) StartStoreSpan(
	ctx context.Context,
	storeName string,
	queryName string,
	kind QueryKind,
) (context.Context, *StoreSpan)

StartStoreSpan starts telemetry around a factory-wrapped generated store method. Generated internal methods mark the returned event through ctx while creating their own child query spans.

type MeshOption 

type MeshOption func(*meshConfig)

MeshOption customizes a declaratively constructed mesh.

func WithLogger 

func WithLogger(logger *slog.Logger) MeshOption

WithLogger configures optional structured logging for routed queries. A nil logger disables logging.

func WithMeterProvider 

func WithMeterProvider(provider metric.MeterProvider) MeshOption

WithMeterProvider configures the provider used for routed query metrics. A nil provider uses the global OpenTelemetry meter provider.

func WithReplicaSet 

func WithReplicaSet(name, primaryDSN string, replicaDSNs ...string) MeshOption

WithReplicaSet registers a named primary and its optional read replicas. Repeated calls append replica sets in call order.

func WithTracerProvider 

func WithTracerProvider(provider trace.TracerProvider) MeshOption

WithTracerProvider configures the provider used for routed query spans. A nil provider uses the global OpenTelemetry tracer provider.

func WithVShardMapping 

func WithVShardMapping(
	mainReplicaSet string,
	vshards []uint64,
	mirrorReplicaSets ...string,
) MeshOption

WithVShardMapping maps virtual shards to a main replica set and optional ordered write mirrors. Repeated calls append mappings in call order.

type Mirrorable 

type Mirrorable[W any] interface {
	// WithMirrors returns a copy that also writes to the supplied mirrors.
	WithMirrors(...W) W
}

Mirrorable is implemented by generated primary-capable query wrappers. WithMirrors must return a new value and leave the receiver unchanged.

type Node 

type Node[R any, W Mirrorable[W]] struct {
	// contains filtered or unexported fields
}

Node contains the read-only and primary-capable views of one database connection. ReplicaSet exposes only Reader for replicas and Writer for the primary, preventing writes from accidentally being routed to replicas.

func NewNode 

func NewNode[R any, W Mirrorable[W]](reader R, writer W) Node[R, W]

NewNode creates a database node from its read-only and primary-capable views.

func (Node[R, W]) Reader 

func (n Node[R, W]) Reader() R

Reader returns the node's read-only query view.

func (Node[R, W]) Writer 

func (n Node[R, W]) Writer() W

Writer returns the node's primary-capable query view.

type NodeFactory 

type NodeFactory[R any, W Mirrorable[W]] func(context.Context, string) (Node[R, W], error)

NodeFactory opens the database node identified by a DSN. Nodes and their underlying pools remain caller-owned; pgmesh does not close them.

type NodeRole added in v0.0.5 

type NodeRole string

NodeRole identifies the database node selected for a physical query. 

const (
	// NodeRolePrimary identifies a replica set's writable primary node.
	NodeRolePrimary NodeRole = "primary"
	// NodeRoleReadReplica identifies a read-only replica node.
	NodeRoleReadReplica NodeRole = "read_replica"
	// NodeRoleTransaction identifies an externally supplied transaction whose
	// underlying database node is not observable by pgmesh.
	NodeRoleTransaction NodeRole = "transaction"
)

type PhysicalQuerySpan added in v0.0.5 

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

PhysicalQuerySpan records one database execution on one resolved node.

func (*PhysicalQuerySpan) End added in v0.0.5 

func (s *PhysicalQuerySpan) End(err error)

End records one physical database-query metric and span.

type QueryKind 

type QueryKind string

QueryKind classifies a routed query as a read or write. 

const (
	// QueryKindRead identifies a read query.
	QueryKindRead QueryKind = "read"
	// QueryKindWrite identifies a write query.
	QueryKindWrite QueryKind = "write"
)

Query kinds recorded by generated routed query methods.

type QuerySpan 

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

QuerySpan records tracing, metrics, and logging for one routed query. The generated store calls End exactly once; callers using StartSpan directly must do the same.

func (*QuerySpan) End 

func (s *QuerySpan) End(err error)

End records the logical operation metric and span.

func (*QuerySpan) SetMultiRoute 

func (s *QuerySpan) SetMultiRoute(mode RouteMode, shardCount int)

SetMultiRoute records the number of physical shards resolved by an operation that can fan out. A single resolved shard retains normal single-route scope. Each database execution must also use StartQuerySpan.

func (*QuerySpan) SetRoute 

func (s *QuerySpan) SetRoute(mode RouteMode)

SetRoute records that one logical operation targets a single physical shard.

func (*QuerySpan) StartQuerySpan added in v0.0.5 

func (s *QuerySpan) StartQuerySpan(
	ctx context.Context,
	route RouteMetadata,
	mode RouteMode,
) (context.Context, *PhysicalQuerySpan)

StartQuerySpan starts one physical database-query child of this operation. The returned context must be passed to the selected route's Target.

type ReplicaSet 

type ReplicaSet[R any, W Mirrorable[W]] struct {
	// contains filtered or unexported fields
}

ReplicaSet represents one physical shard. Reads are balanced across replica readers, while writes always use the primary writer and its configured synchronous mirrors. ReplicaSet routing is safe for concurrent use when the configured nodes and writer values are safe for concurrent use.

func NewReplicaSet 

func NewReplicaSet[R any, W Mirrorable[W]](
	name string,
	primary Node[R, W],
	replicas []Node[R, W],
) *ReplicaSet[R, W]

NewReplicaSet creates a physical replica set. If replicas is empty, reads fall back to the primary node.

func (*ReplicaSet[R, W]) Name 

func (s *ReplicaSet[R, W]) Name() string

Name returns the replica set's topology name.

func (*ReplicaSet[R, W]) Read 

func (s *ReplicaSet[R, W]) Read() R

Read returns the next read view selected by round-robin balancing.

func (*ReplicaSet[R, W]) WithWriteMirrors 

func (s *ReplicaSet[R, W]) WithWriteMirrors(writes ...W) *ReplicaSet[R, W]

WithWriteMirrors returns a copy with writes appended to its synchronous mirrors. It does not mutate the receiver, and Write passes mirrors to the writer in the same order in which they were configured.

func (*ReplicaSet[R, W]) Write 

func (s *ReplicaSet[R, W]) Write() W

Write returns the primary write view configured with synchronous mirrors.

func (*ReplicaSet[R, W]) WriteMirrorCount 

func (s *ReplicaSet[R, W]) WriteMirrorCount() int

WriteMirrorCount returns the number of synchronous write mirrors.

type Route added in v0.0.5 

type Route[T any] struct {
	Target       T
	VirtualShard uint64
	Shard        string
	Node         string
	Role         NodeRole
}

Route is one fully resolved physical database target. Node is stable within a replica set: "primary" for its primary and "replica-N" for the Nth configured read replica.

func (Route[T]) Metadata added in v0.0.5 

func (r Route[T]) Metadata() RouteMetadata

Metadata returns the target identity used by telemetry.

type RouteMetadata added in v0.0.5 

type RouteMetadata struct {
	VirtualShard    uint64
	HasVirtualShard bool
	Shard           string
	Node            string
	Role            NodeRole
}

RouteMetadata is the identity of a resolved database target.

func (RouteMetadata) WithoutVirtualShard added in v0.0.5 

func (r RouteMetadata) WithoutVirtualShard() RouteMetadata

WithoutVirtualShard removes a virtual-shard attribution when one physical query combines work from several virtual shards or scans a physical shard.

type RouteMode 

type RouteMode string

RouteMode describes the database path selected for a routed query. 

const (
	// RouteModeRead indicates a read routed through the replica load balancer.
	RouteModeRead RouteMode = "read"
	// RouteModePrimary indicates a read or write routed directly to the primary.
	RouteModePrimary RouteMode = "primary"
	// RouteModeTransaction indicates a query executed on an explicit transaction.
	RouteModeTransaction RouteMode = "transaction"
	// RouteModeUnresolved indicates that routing failed before selecting a path.
	RouteModeUnresolved RouteMode = "unresolved"
)

Route modes recorded after a query resolves to a shard.

type RouteScope added in v0.0.5 

type RouteScope string

RouteScope classifies the fan-out of one logical operation. 

const (
	// RouteScopeSingle identifies an operation targeting at most one shard.
	RouteScopeSingle RouteScope = "single"
	// RouteScopeFanout identifies an operation targeting multiple shards.
	RouteScopeFanout RouteScope = "fanout"
	// RouteScopeUnresolved indicates that routing did not resolve a scope.
	RouteScopeUnresolved RouteScope = "unresolved"
)

type Shard 

type Shard[R any, W Mirrorable[W]] struct {
	*ReplicaSet[R, W]
	// contains filtered or unexported fields
}

Shard is a routed virtual shard linked to a physical replica set.

func (*Shard[R, W]) ReadRoute added in v0.0.5 

func (s *Shard[R, W]) ReadRoute() Route[R]

ReadRoute selects and describes the next read target. Callers must execute the query through the returned Target so selection and telemetry agree.

func (*Shard[R, W]) VShardIndex 

func (s *Shard[R, W]) VShardIndex() uint64

VShardIndex returns the virtual shard index used to select this shard.

func (*Shard[R, W]) WriteRoute added in v0.0.5 

func (s *Shard[R, W]) WriteRoute() Route[W]

WriteRoute selects and describes the primary write target.

type ShardHasher 

type ShardHasher[SK any] interface {
	// Hash returns the virtual shard index for key.
	Hash(SK) uint64
}

ShardHasher maps an application shard key to a virtual shard index.

func ConstantShardHashFor 

func ConstantShardHashFor[SK any](vshard uint64) ShardHasher[SK]

ConstantShardHashFor returns a hasher that always selects vshard.

func ModularShardHashFor 

func ModularShardHashFor[SK IntShardKey](numVShards uint64) ShardHasher[SK]

ModularShardHashFor returns a hasher that maps integer keys modulo numVShards. Signed keys use Euclidean modulo, so negative values map into the same [0, numVShards) range without overflowing at the minimum integer value. Named integer types are supported. It panics if numVShards is zero.

type StoreSpan added in v0.0.3 

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

StoreSpan records tracing, metrics, and logging around one factory-wrapped generated store method. The generated wrapper calls End exactly once.

func (*StoreSpan) End added in v0.0.3 

func (s *StoreSpan) End(err error)

End records metrics and a debug log, records err if present, then ends the factory-wrapped store span. The configured providers and logger remain caller-owned.

Source Files

View all Source files

Directories

Path Synopsis
cmd
sqlc-gen-store command
examples
01-single-database command
02-read-write-split command
03-sharded-read-write command
04-mirrors-and-transactions command
05-async-copy-batching command
06-cache-aside command
07-multi-shard-queries command
internal/one
internal/sharded
Package sqlcplugin generates pgmesh query wrappers from sqlc metadata.
 Package sqlcplugin generates pgmesh query wrappers from sqlc metadata.
tests
command_shapes/internal
command_shapes/store
internal/testdb
same_package
separate_package/internal
separate_package/store

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.