rmap: fix SetAndWait race (#62)

* rmap: fix SetAndWait race

* pool: make jump hash concurrency-safe

Author: raphael

pool/node.go

@@ -72,7 +72,8 @@ type (
 
 	// jumpHash implement Jump Consistent Hash.
 	jumpHash struct {
-		h hash.Hash64
+		mu sync.Mutex
+		h  hash.Hash64
 	}
 )
 
@@ -245,7 +246,7 @@ func AddNode(ctx context.Context, poolName string, rdb *redis.Client, opts ...No
 		workerTTL:          o.workerTTL,
 		workerShutdownTTL:  o.workerShutdownTTL,
 		ackGracePeriod:     o.ackGracePeriod,
-		h:                  jumpHash{crc64.New(crc64.MakeTable(crc64.ECMA))},
+		h:                  &jumpHash{h: crc64.New(crc64.MakeTable(crc64.ECMA))},
 		stop:               make(chan struct{}),
 		closed:             closed,
 		rdb:                rdb,
@@ -1404,13 +1405,18 @@ func (node *Node) maps() []*rmap.Map {
 
 // Hash implements the Jump Consistent Hash algorithm.
 // See https://arxiv.org/ftp/arxiv/papers/1406/1406.2294.pdf for details.
-func (jh jumpHash) Hash(key string, numBuckets int64) int64 {
+func (jh *jumpHash) Hash(key string, numBuckets int64) int64 {
 	var b int64 = -1
 	var j int64
 
+	jh.mu.Lock()
 	jh.h.Reset()
-	io.WriteString(jh.h, key) // nolint: errcheck
+	_, err := io.WriteString(jh.h, key)
 	sum := jh.h.Sum64()
+	jh.mu.Unlock()
+	if err != nil {
+		panic(fmt.Errorf("jumpHash: write key: %w", err))
+	}
 
 	for j < numBuckets {
 		b = j

rmap/map.go

@@ -258,9 +258,8 @@ func (sm *Map) SetAndWait(ctx context.Context, key, value string) (string, error
 		cancel: cancel,
 	}
 
-	// First mark the channel as closing before removing from waiters
 	defer func() {
-		// Cancel first to prevent new sends
+		// Cancel first so notifiers prefer to drop any late sends.
 		cancel()
 
 		// Remove waiter under lock
@@ -280,9 +279,6 @@ func (sm *Map) SetAndWait(ctx context.Context, key, value string) (string, error
 			}
 		}
 		sm.wlock.Unlock()
-
-		// Now safe to close channel as no more sends will occur
-		close(notifyCh)
 	}()
 
 	// Prepare new waiters list under lock
@@ -681,6 +677,8 @@ func (sm *Map) run() {
 							case <-waiter.ctx.Done():
 								// Waiter was cancelled or timed out
 								continue
+							default:
+								// Non-blocking to avoid stalling the map loop if the waiter is no longer receiving.
 							}
 						}
 					}