cre-1601: hash ring pure function refactor for both storage and observation

Author: mchain0

pkg/workflows/ring/plugin.go

@@ -191,9 +191,10 @@ func (p *Plugin) Outcome(_ context.Context, outctx ocr3types.OutcomeContext, _ t
 	}
 
 	// Use deterministic hashing to assign workflows to shards
+	// This must be a pure function of consensus-derived data to avoid protocol failures
 	routes := make(map[string]*pb.WorkflowRoute)
 	for _, wfID := range allWorkflows {
-		assignedShard := p.store.GetShardForWorkflow(wfID)
+		assignedShard := getShardForWorkflow(wfID, healthyShardCount)
 		routes[wfID] = &pb.WorkflowRoute{
 			Shard: assignedShard,
 		}

pkg/workflows/ring/store.go

@@ -2,34 +2,15 @@ package ring
 
 import (
 	"slices"
-	"strconv"
 	"sync"
-
-	"github.com/buraksezer/consistent"
-	"github.com/cespare/xxhash/v2"
 )
 
-// xxhashHasher implements the consistent.Hasher interface using xxhash
-type xxhashHasher struct{}
-
-func (h xxhashHasher) Sum64(data []byte) uint64 {
-	return xxhash.Sum64(data)
-}
-
-// ShardMember implements consistent.Member for shard IDs
-type ShardMember string
-
-func (m ShardMember) String() string {
-	return string(m)
-}
-
 // Store manages shard routing state and workflow mappings
 type Store struct {
-	routingState   map[string]uint32      // workflow_id -> shard_id
-	shardHealth    map[uint32]bool        // shard_id -> is_healthy
-	consistentHash *consistent.Consistent // Consistent hash ring for routing
-	healthyShards  []uint32               // Sorted list of healthy shards
-	mu             sync.Mutex
+	routingState  map[string]uint32 // workflow_id -> shard_id
+	shardHealth   map[uint32]bool   // shard_id -> is_healthy
+	healthyShards []uint32          // Sorted list of healthy shards
+	mu            sync.Mutex
 }
 
 func NewStore() *Store {
@@ -41,29 +22,15 @@ func NewStore() *Store {
 	}
 }
 
-// consistentHashConfig returns the configuration for consistent hashing
-// Matches prototype: PartitionCount=997 (prime), ReplicationFactor=50, Load=1.1
-func consistentHashConfig() consistent.Config {
-	return consistent.Config{
-		PartitionCount:    997, // Prime number for better distribution
-		ReplicationFactor: 50,  // Number of replicas per node
-		Load:              1.1, // Load factor for bounded loads
-		Hasher:            xxhashHasher{},
-	}
-}
-
-// updateConsistentHash rebuilds the consistent hash ring based on healthy shards
-func (s *Store) updateConsistentHash() {
+// updateHealthyShards rebuilds the sorted list of healthy shards
+func (s *Store) updateHealthyShards() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	// Get list of healthy shards and create members
-	healthyMembers := make([]consistent.Member, 0)
 	s.healthyShards = make([]uint32, 0)
 
 	for shardID, healthy := range s.shardHealth {
 		if healthy {
-			healthyMembers = append(healthyMembers, ShardMember(strconv.FormatUint(uint64(shardID), 10)))
 			s.healthyShards = append(s.healthyShards, shardID)
 		}
 	}
@@ -72,39 +39,18 @@ func (s *Store) updateConsistentHash() {
 	slices.Sort(s.healthyShards)
 
 	// If no healthy shards, add shard 0 as fallback
-	if len(healthyMembers) == 0 {
-		healthyMembers = append(healthyMembers, ShardMember("0"))
+	if len(s.healthyShards) == 0 {
 		s.healthyShards = []uint32{0}
 	}
-
-	// Create consistent hash ring
-	s.consistentHash = consistent.New(healthyMembers, consistentHashConfig())
 }
 
 // GetShardForWorkflow deterministically assigns a workflow to a shard using consistent hashing.
-// The assignment uses the same algorithm as the prototype: xxhash + consistent hashing ring.
 func (s *Store) GetShardForWorkflow(workflowID string) uint32 {
 	s.mu.Lock()
-	defer s.mu.Unlock()
-
-	if s.consistentHash == nil {
-		// Fallback if hash ring not initialized
-		return 0
-	}
-
-	// Use consistent hashing to find the member for this workflow
-	member := s.consistentHash.LocateKey([]byte(workflowID))
-	if member == nil {
-		return 0
-	}
-
-	// Parse shard ID from member name
-	shardID, err := strconv.ParseUint(member.String(), 10, 32)
-	if err != nil {
-		return 0
-	}
+	shardCount := uint32(len(s.healthyShards))
+	s.mu.Unlock()
 
-	return uint32(shardID)
+	return getShardForWorkflow(workflowID, shardCount)
 }
 
 func (s *Store) SetShardForWorkflow(workflowID string, shardID uint32) {
@@ -128,8 +74,8 @@ func (s *Store) SetShardHealth(shardID uint32, healthy bool) {
 	s.shardHealth[shardID] = healthy
 	s.mu.Unlock()
 
-	// Rebuild consistent hash ring when shard health changes
-	s.updateConsistentHash()
+	// Rebuild healthy shards list when shard health changes
+	s.updateHealthyShards()
 }
 
 func (s *Store) SetAllShardHealth(health map[uint32]bool) {
@@ -140,8 +86,8 @@ func (s *Store) SetAllShardHealth(health map[uint32]bool) {
 	}
 	s.mu.Unlock()
 
-	// Rebuild consistent hash ring
-	s.updateConsistentHash()
+	// Rebuild healthy shards list
+	s.updateHealthyShards()
 }
 
 func (s *Store) GetAllRoutingState() map[string]uint32 {

pkg/workflows/ring/utils.go

@@ -1,10 +1,65 @@
 package ring
 
-import "slices"
+import (
+	"slices"
+	"strconv"
+
+	"github.com/buraksezer/consistent"
+	"github.com/cespare/xxhash/v2"
+)
 
 func uniqueSorted(s []string) []string {
 	result := slices.Clone(s)
 	slices.Sort(result)
 	return slices.Compact(result)
 }
 
+type xxhashHasher struct{}
+
+func (h xxhashHasher) Sum64(data []byte) uint64 {
+	return xxhash.Sum64(data)
+}
+
+type ShardMember string
+
+func (m ShardMember) String() string {
+	return string(m)
+}
+
+func consistentHashConfig() consistent.Config {
+	return consistent.Config{
+		PartitionCount:    997, // Prime number for better distribution
+		ReplicationFactor: 50,  // Number of replicas per node
+		Load:              1.1, // Load factor for bounded loads
+		Hasher:            xxhashHasher{},
+	}
+}
+
+func getShardForWorkflow(workflowID string, shardCount uint32) uint32 {
+	if shardCount == 0 {
+		return 0
+	}
+
+	// Create members for shards 0 to shardCount-1
+	members := make([]consistent.Member, shardCount)
+	for i := uint32(0); i < shardCount; i++ {
+		members[i] = ShardMember(strconv.FormatUint(uint64(i), 10))
+	}
+
+	// Create consistent hash ring
+	ring := consistent.New(members, consistentHashConfig())
+
+	// Use consistent hashing to find the member for this workflow
+	member := ring.LocateKey([]byte(workflowID))
+	if member == nil {
+		return 0
+	}
+
+	// Parse shard ID from member name
+	shardID, err := strconv.ParseUint(member.String(), 10, 32)
+	if err != nil {
+		return 0
+	}
+
+	return uint32(shardID)
+}