Refactor/unify shard cluster

CHANGELOG.md

@@ -2,6 +2,37 @@
 
 All notable changes to this project will be documented in this file.
 
+## v0.3.0
+
+This release reorganizes the cluster and shard APIs under a common topology namespace and simplifies several sharding
+contracts.
+
+### Changed
+
+* **Topology Package Layout:** Moved cluster and shard packages to `topology/cluster` and `topology/shard`, with shard
+  resolvers under `topology/shard/resolver`.
+* **Cluster Identity:** Cluster IDs are now required when creating a `cluster.Cluster`.
+* **Shard Topology Construction:** Simplified topology creation from `shard.NewTopology([]shard.Config{...})` to
+  `shard.NewTopology(clusters...)`.
+* **Rendezvous Resolver:** Renamed `HashResolver` and `NewHash` to `RendezvousResolver` and `NewRendezvous`, making the
+  routing algorithm explicit while preserving the existing rendezvous placement contract.
+* **Custom Resolvers:** Simplified custom resolver callbacks to map a key directly to `shard.ID` without receiving the
+  topology on every call.
+* **Cross-Shard Operations:** `ForEachShard` now returns callback and cancellation failures through its function error
+  while preserving detailed per-shard results.
+
+### Fixed
+
+* **Rendezvous Portability:** Fixed length validation in rendezvous routing so the resolver also compiles correctly on
+  32-bit architectures.
+
+### Removed
+
+* **Shard Configuration Layer:** Removed `shard.Config`; shard topologies are now created directly from clusters.
+* **Generic Hash API:** Removed the `HashResolver` and `NewHash` names in favor of the explicit rendezvous API.
+
+---
+
 ## v0.2.0
 
 Initial production release of `xpg`, built around `pgx` with PostgreSQL transaction helpers, primary/replica clustering,

README.md

@@ -26,10 +26,10 @@ connection management, routing, and common production workflows.
 * **Error Classification:** Classification of PostgreSQL constraint, transaction, cancellation, connection, and other
   common database errors.
 * **Advisory Locking:** Transaction-level advisory locks for coordinating concurrent database operations.
-* **Primary/Replica Routing:** Explicit read policies, replica selection, primary fallback, and read-only transactions
-  across PostgreSQL nodes.
-* **Application-Level Sharding:** Hash, range, time-based, and custom routing with colocation checks, key grouping, and
-  bounded parallel operations across shards.
+* **Primary/Replica Routing:** Logical cluster topologies with explicit read policies, replica selection, primary
+  fallback, and read-only transactions across PostgreSQL nodes.
+* **Application-Level Sharding:** Rendezvous, range, time-based, and custom routing with colocation checks, key
+  grouping, and bounded parallel operations across shards.
 * **Observability:** Structured logging, tracing, pool statistics, and optional OpenTelemetry metrics.
 
 ## Installation
@@ -132,7 +132,7 @@ serialization failures, deadlocks, lock errors, query cancellation, and connecti
 
 ## Clustering
 
-`xpg` groups primary and replica pools into a logical cluster with explicit read routing.
+The `topology/cluster` package groups primary and replica pools into a logical cluster with explicit read routing.
 
 <!-- @formatter:off -->
 ```go
@@ -173,61 +173,60 @@ replica is available. Replica selection is round-robin by default and can be cus
 
 ## Sharding
 
-`xpg` provides application-level sharding with explicit key routing across an immutable shard topology.
+The `topology/shard` package provides application-level sharding with explicit key routing across an immutable shard
+topology. Routing strategies live under `topology/shard/resolver`.
 
 <!-- @formatter:off -->
 ```go
-topology, err := shard.NewTopology([]shard.Config{
-    {Cluster: shardA},
-    {Cluster: shardB},
-})
+topology, err := shard.NewTopology(shardA, shardB)
 if err != nil {
-    panic(err)
+	panic(err)
 }
 defer topology.Close()
 
 // Partition user IDs into shard ranges.
 users, err := resolver.NewRange(
-    topology,
-    []resolver.Range[uint64]{
-        {Start: 0, End: 100, ShardID: "shard-a"},
-        {Start: 100, End: 200, ShardID: "shard-b"},
-    },
+	topology,
+	[]resolver.Range[uint64]{
+		{Start: 0, End: 100, ShardID: "shard-a"},
+		{Start: 100, End: 200, ShardID: "shard-b"},
+	},
 )
 if err != nil {
-    panic(err)
+	panic(err)
 }
 
 // Resolve the target shard.
-shard, err := users.Resolve(userID)
+targetShard, err := users.Resolve(userID)
 if err != nil {
-    panic(err)
+	panic(err)
 }
 
 // Write to the shard primary.
-primaryPool := shard.Primary()
+primaryPool := targetShard.Primary()
 
 _, err = primaryPool.Exec(ctx, "UPDATE users SET active = true WHERE id = $1", userID)
 if err != nil {
-    panic(err)
+	panic(err)
 }
 
 // Read from the same shard using the selected read policy.
-readPool, err := shard.ReadPool(ctx, cluster.ReadReplicaPreferred)
+readPool, err := targetShard.ReadPool(ctx, cluster.ReadReplicaPreferred)
 if err != nil {
-    panic(err)
+	panic(err)
 }
 
 var active bool
 err = readPool.QueryRow(ctx, "SELECT active FROM users WHERE id = $1", userID).Scan(&active)
 if err != nil {
-    panic(err)
+	panic(err)
 }
+
 ```
 <!-- @formatter:on -->
 
 Built-in routing strategies include rendezvous hashing, ordered ranges, time ranges, and custom resolvers. Sharding
-utilities cover key colocation, grouping by shard, and parallel operations across shards.
+utilities cover key colocation, grouping by shard, and bounded parallel operations across shards.
 
 ## Examples
 

examples/basic/main.go

@@ -104,11 +104,7 @@ func upsertUsers(ctx context.Context, pool *xpg.Pool) (int64, error) {
 	return tag.RowsAffected(), nil
 }
 
-func loadUser(
-	ctx context.Context,
-	pool *xpg.Pool,
-	userID int64,
-) (user, error) {
+func loadUser(ctx context.Context, pool *xpg.Pool, userID int64) (user, error) {
 	var selected user
 
 	err := pool.QueryRow(
@@ -134,10 +130,7 @@ func loadUser(
 	return selected, nil
 }
 
-func listActiveUsers(
-	ctx context.Context,
-	pool *xpg.Pool,
-) ([]user, error) {
+func listActiveUsers(ctx context.Context, pool *xpg.Pool) ([]user, error) {
 	rows, err := pool.Query(
 		ctx,
 		`SELECT

examples/cluster/main.go

@@ -6,7 +6,7 @@ import (
 	"log"
 
 	"github.com/jackc/pgx/v5"
-	"github.com/mkbeh/xpg/cluster"
+	"github.com/mkbeh/xpg/topology/cluster"
 )
 
 type nodeInfo struct {

examples/cluster/setup.go

@@ -4,9 +4,10 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"slices"
 
 	"github.com/mkbeh/xpg"
-	"github.com/mkbeh/xpg/cluster"
+	"github.com/mkbeh/xpg/topology/cluster"
 )
 
 const (
@@ -16,50 +17,64 @@ const (
 )
 
 func openCluster(ctx context.Context) (*cluster.Cluster, error) {
-	primary, err := openPool(
-		ctx,
-		environment("XPG_PRIMARY_DATABASE_URL", defaultPrimaryDatabaseURL),
-		"cluster.primary",
-		"primary",
-	)
-	if err != nil {
-		return nil, fmt.Errorf("open primary pool: %w", err)
+	type poolConfig struct {
+		databaseURL string
+		name        string
+		role        string
 	}
 
-	replicaOne, err := openPool(
-		ctx,
-		environment("XPG_REPLICA_ONE_DATABASE_URL", defaultReplicaOneURL),
-		"cluster.replica-one",
-		"replica",
-	)
-	if err != nil {
-		primary.Close()
-
-		return nil, fmt.Errorf("open replica-one pool: %w", err)
+	configs := []poolConfig{
+		{
+			databaseURL: environment(
+				"XPG_PRIMARY_DATABASE_URL",
+				defaultPrimaryDatabaseURL,
+			),
+			name: "cluster.primary",
+			role: "primary",
+		},
+		{
+			databaseURL: environment(
+				"XPG_REPLICA_ONE_DATABASE_URL",
+				defaultReplicaOneURL,
+			),
+			name: "cluster.replica-one",
+			role: "replica",
+		},
+		{
+			databaseURL: environment(
+				"XPG_REPLICA_TWO_DATABASE_URL",
+				defaultReplicaTwoURL,
+			),
+			name: "cluster.replica-two",
+			role: "replica",
+		},
 	}
 
-	replicaTwo, err := openPool(
-		ctx,
-		environment("XPG_REPLICA_TWO_DATABASE_URL", defaultReplicaTwoURL),
-		"cluster.replica-two",
-		"replica",
-	)
-	if err != nil {
-		replicaOne.Close()
-		primary.Close()
+	pools := make([]*xpg.Pool, 0, len(configs))
+
+	for _, config := range configs {
+		pool, err := openPool(
+			ctx,
+			config.databaseURL,
+			config.name,
+			config.role,
+		)
+		if err != nil {
+			closePools(pools)
+
+			return nil, fmt.Errorf("open %s pool: %w", config.name, err)
+		}
 
-		return nil, fmt.Errorf("open replica-two pool: %w", err)
+		pools = append(pools, pool)
 	}
 
 	dbCluster, err := cluster.New(cluster.Config{
 		ID:       "cluster-example",
-		Primary:  primary,
-		Replicas: []*xpg.Pool{replicaOne, replicaTwo},
+		Primary:  pools[0],
+		Replicas: pools[1:],
 	})
 	if err != nil {
-		replicaTwo.Close()
-		replicaOne.Close()
-		primary.Close()
+		closePools(pools)
 
 		return nil, fmt.Errorf("create cluster: %w", err)
 	}
@@ -87,6 +102,12 @@ func openPool(ctx context.Context, databaseURL, name, role string) (*xpg.Pool, e
 	return pool, nil
 }
 
+func closePools(pools []*xpg.Pool) {
+	for _, pool := range slices.Backward(pools) {
+		pool.Close()
+	}
+}
+
 func environment(key, fallback string) string {
 	if value := os.Getenv(key); value != "" {
 		return value

examples/shard/main.go

@@ -5,8 +5,8 @@ import (
 	"fmt"
 	"log"
 
-	"github.com/mkbeh/xpg/shard"
-	"github.com/mkbeh/xpg/shard/resolver"
+	"github.com/mkbeh/xpg/topology/shard"
+	"github.com/mkbeh/xpg/topology/shard/resolver"
 )
 
 const (
@@ -33,6 +33,9 @@ func run(ctx context.Context) error {
 	}
 	defer topology.Close()
 
+	// A resolver describes how one dataset is distributed across the topology.
+	// NewRange resolves shard IDs once and stores the resulting Shard handles in
+	// an immutable routing table used by subsequent lookups.
 	userResolver, err := resolver.NewRange(
 		topology,
 		[]resolver.Range[uint64]{
@@ -60,14 +63,14 @@ func run(ctx context.Context) error {
 	fmt.Println("range routing:")
 
 	for _, current := range users {
-		resolved, err := userResolver.Resolve(current.ID)
+		targetShard, err := userResolver.Resolve(current.ID)
 		if err != nil {
 			return fmt.Errorf("resolve user %d: %w", current.ID, err)
 		}
 
-		primary := resolved.Primary()
+		primary := targetShard.Primary()
 		if primary == nil {
-			return fmt.Errorf("shard %q has no primary", resolved.ID())
+			return fmt.Errorf("shard %q has no primary", targetShard.ID())
 		}
 
 		if _, err := primary.Exec(
@@ -85,7 +88,7 @@ func run(ctx context.Context) error {
 		fmt.Printf(
 			"- user_id=%d shard=%s pool=%s\n",
 			current.ID,
-			resolved.ID(),
+			targetShard.ID(),
 			primary.Name(),
 		)
 	}