pgmesh

package module
v0.0.4 Latest Latest
Warning

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

 Go to latest Published: Jul 29, 2026 License: MIT Imports: 14 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, grouped list lookups, and physical-shard grouping for COPY FROM;
  • 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 progressive examples from one database through replicas, sharding, write mirrors, and transactions.

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"
	// AttributeReplicaSet identifies the selected physical replica set.
	AttributeReplicaSet = "pgmesh.route.replica_set"
	// AttributeRouteMode identifies the database path selected for a query.
	AttributeRouteMode = "pgmesh.route.mode"
	// AttributeInternalStoreExecuted reports whether a factory wrapper entered
	// the generated internal store implementation.
	AttributeInternalStoreExecuted = "pgmesh.store.internal_executed"
)

OpenTelemetry attribute keys recorded on store and routed query telemetry.

View Source 
const MetricQueryDuration = "pgmesh.query.duration"

MetricQueryDuration is the OpenTelemetry histogram of routed query durations in seconds. Its count also reports completed query throughput. The configured MeterProvider owns exporting and shutdown; pgmesh never shuts it down.

View Source 
const MetricStoreDuration = "pgmesh.store.duration"

MetricStoreDuration is the OpenTelemetry histogram of factory-wrapped store method 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")
)

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 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]) 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]) StartSpan 

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

StartSpan starts telemetry for a routed query and returns the span context so database instrumentation can create child spans.

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 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 metrics and a debug log, records err if present, then ends the routed query span. The configured providers and logger remain caller-owned.

func (*QuerySpan) SetMultiRoute 

func (s *QuerySpan) SetMultiRoute(mode RouteMode)

SetMultiRoute records the routing mode for one logical operation targeting zero or more physical replica sets. It deliberately omits a virtual-shard index and replica-set name because no single value represents the operation.

func (*QuerySpan) SetRoute 

func (s *QuerySpan) SetRoute(
	vshard uint64,
	replicaSet string,
	mode RouteMode,
)

SetRoute records the selected virtual shard for debug logging and the bounded physical-route attributes used by tracing and metrics.

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 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"
)

Route modes recorded after a query resolves to a shard.

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]) VShardIndex 

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

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

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
internal/one
internal/sharded
Package sqlcplugin generates pgmesh query wrappers from sqlc metadata.
 Package sqlcplugin generates pgmesh query wrappers from sqlc metadata.
tests
generate/commands/internal
generate/commands/store
generate/same_package
generate/separate_package/internal
generate/separate_package/store
integration/testdb

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.