perf: route remote HTTP puts to owner

coordinator/coordinator.go

@@ -340,6 +340,17 @@ func (c *Coordinator) Route(key string) (pb.CacheServiceClient, error) {
 	return c.router.Route(key)
 }
 
+// Resolve returns the current owner for a key without acquiring a peer client.
+func (c *Coordinator) Resolve(key string) (*ring.NodeInfo, error) {
+	return c.router.Resolve(key)
+}
+
+// RouteToNode returns a client for an already-resolved owner while preserving
+// the router's connection retry, health, and circuit-breaker behavior.
+func (c *Coordinator) RouteToNode(nodeID string) (pb.CacheServiceClient, error) {
+	return c.router.RouteToNode(nodeID)
+}
+
 // GetNodeForKey returns the node for the given key
 func (c *Coordinator) GetNodeForKey(key string) (*ring.NodeInfo, error) {
 	return c.ringManager.GetNode(key)

coordinator/interceptors.go

@@ -59,9 +59,12 @@ type ResponseMetadata struct {
 	ForwardedBy string
 }
 
-// ExtractRequestMetadata extracts routing metadata from the incoming context
+// ExtractRequestMetadata extracts routing metadata from incoming or outgoing context.
 func ExtractRequestMetadata(ctx context.Context) RequestMetadata {
 	md, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
+		md, ok = metadata.FromOutgoingContext(ctx)
+	}
 	if !ok {
 		return RequestMetadata{}
 	}
@@ -97,12 +100,17 @@ func ExtractRequestMetadata(ctx context.Context) RequestMetadata {
 
 // AttachForwardingMetadata attaches forwarding metadata to outgoing context
 func AttachForwardingMetadata(ctx context.Context, rm RequestMetadata) context.Context {
-	return metadata.AppendToOutgoingContext(ctx,
-		MetadataKeyRingEpoch, strconv.FormatUint(rm.RingEpoch, 10),
-		MetadataKeyHop, strconv.Itoa(rm.HopCount),
-		MetadataKeyForwarded, strconv.FormatBool(rm.Forwarded),
-		MetadataKeyOrigin, rm.OriginNode,
-	)
+	md, ok := metadata.FromOutgoingContext(ctx)
+	if !ok {
+		md = metadata.MD{}
+	} else {
+		md = md.Copy()
+	}
+	md.Set(MetadataKeyRingEpoch, strconv.FormatUint(rm.RingEpoch, 10))
+	md.Set(MetadataKeyHop, strconv.Itoa(rm.HopCount))
+	md.Set(MetadataKeyForwarded, strconv.FormatBool(rm.Forwarded))
+	md.Set(MetadataKeyOrigin, rm.OriginNode)
+	return metadata.NewOutgoingContext(ctx, md)
 }
 
 // CheckHopCount validates that the hop count hasn't exceeded MaxHops.

coordinator/interceptors_test.go

@@ -105,6 +105,26 @@ func TestAttachForwardingMetadata(t *testing.T) {
 	assert.Equal(t, []string{"node1"}, md.Get(MetadataKeyOrigin))
 }
 
+func TestIncrementHopCountUsesOutgoingMetadata(t *testing.T) {
+	ctx := metadata.NewOutgoingContext(context.Background(), metadata.Pairs(
+		MetadataKeyRingEpoch, "42",
+		MetadataKeyHop, "1",
+		MetadataKeyOrigin, "origin",
+		"authorization", "bearer",
+	))
+
+	forwarded, err := IncrementHopCount(ctx, "gateway")
+	require.NoError(t, err)
+
+	md, ok := metadata.FromOutgoingContext(forwarded)
+	require.True(t, ok)
+	assert.Equal(t, []string{"42"}, md.Get(MetadataKeyRingEpoch))
+	assert.Equal(t, []string{"2"}, md.Get(MetadataKeyHop))
+	assert.Equal(t, []string{"true"}, md.Get(MetadataKeyForwarded))
+	assert.Equal(t, []string{"origin"}, md.Get(MetadataKeyOrigin))
+	assert.Equal(t, []string{"bearer"}, md.Get("authorization"))
+}
+
 func TestCheckHopCount(t *testing.T) {
 	tests := []struct {
 		name      string

coordinator/router.go

@@ -124,6 +124,13 @@ func NewRouterWithConfig(ring Ring, localID string, config *RouterConfig) *Route
 	}
 }
 
+// Resolve returns the current owner for a key without acquiring a peer client.
+// Keeping ownership lookup separate lets callers that already resolved a key
+// acquire the corresponding cached client without hashing the key again.
+func (r *Router) Resolve(key string) (*ring.NodeInfo, error) {
+	return r.ring.GetNode(key)
+}
+
 // Route returns a client for routing requests for the given key
 // Returns an error if the key should be handled locally (defensive check)
 func (r *Router) Route(key string) (pb.CacheServiceClient, error) {
@@ -134,7 +141,7 @@ func (r *Router) Route(key string) (pb.CacheServiceClient, error) {
 // Returns an error if the key maps to the local node (this should not happen
 // as callers should check IsLocal first, but we check defensively)
 func (r *Router) RouteWithRetry(key string, maxRetries int) (pb.CacheServiceClient, error) {
-	node, err := r.ring.GetNode(key)
+	node, err := r.Resolve(key)
 	if err != nil {
 		metrics.ClusterRouteRequests.WithLabelValues("error").Inc()
 		return nil, err
@@ -153,6 +160,17 @@ func (r *Router) RouteWithRetry(key string, maxRetries int) (pb.CacheServiceClie
 		return nil, NewLocalRoutingError(r.localID, key)
 	}
 
+	return r.routeToNode(node.ID, key, maxRetries)
+}
+
+// RouteToNode returns a client for an already-resolved owner. It uses the same
+// connection retry, health, and circuit-breaker path as Route, but does not
+// resolve the key again.
+func (r *Router) RouteToNode(nodeID string) (pb.CacheServiceClient, error) {
+	return r.routeToNode(nodeID, "", r.config.MaxRetries)
+}
+
+func (r *Router) routeToNode(nodeID, key string, maxRetries int) (pb.CacheServiceClient, error) {
 	var lastErr error
 	backoff := r.config.InitialRetryBackoff
 
@@ -164,21 +182,17 @@ func (r *Router) RouteWithRetry(key string, maxRetries int) (pb.CacheServiceClie
 			time.Sleep(jittered(backoff))
 			backoff = r.calculateBackoff(backoff)
 
-			metrics.ClusterRetryAttempts.WithLabelValues(node.ID).Inc()
+			metrics.ClusterRetryAttempts.WithLabelValues(nodeID).Inc()
 
 			zlog.Debug().
-				Str("node_id", node.ID).
+				Str("node_id", nodeID).
 				Int("attempt", attempt).
 				Dur("backoff", backoff).
 				Msg("Retrying connection after failure")
 		}
 
-		client, err := r.getClient(node.ID)
+		client, err := r.getClient(nodeID)
 		if err == nil {
-			zlog.Debug().
-				Str("node_id", node.ID).
-				Msg("Successfully routed to node")
-
 			metrics.ClusterRouteRequests.WithLabelValues("remote").Inc()
 			return client, nil
 		}
@@ -193,7 +207,7 @@ func (r *Router) RouteWithRetry(key string, maxRetries int) (pb.CacheServiceClie
 
 	metrics.ClusterRouteRequests.WithLabelValues("error").Inc()
 	metrics.ClusterRoutingErrors.WithLabelValues("max_retries_exceeded").Inc()
-	return nil, NewMaxRetriesExceededError(node.ID, key, maxRetries+1, lastErr)
+	return nil, NewMaxRetriesExceededError(nodeID, key, maxRetries+1, lastErr)
 }
 
 // getClient returns a client for the given node, creating one if necessary
@@ -291,6 +305,9 @@ func (r *Router) getClient(nodeID string) (pb.CacheServiceClient, error) {
 		Str("node_id", nodeID).
 		Str("address", nodeAddr).
 		Msg("Created connection to node")
+	zlog.Debug().
+		Str("node_id", nodeID).
+		Msg("Successfully routed to node")
 
 	return client, nil
 }

coordinator/router_test.go

@@ -309,6 +309,15 @@ func TestRouter_BasicRouting(t *testing.T) {
 	client, err = router.Route("remote-key")
 	assert.NoError(t, err)
 	assert.NotNil(t, client, "Remote routing should return a client")
+
+	// An already-resolved owner should reuse the same client acquisition path
+	// without resolving the key again.
+	owner, err := router.Resolve("remote-key")
+	require.NoError(t, err)
+	assert.Equal(t, "remote-node", owner.ID)
+	resolvedClient, err := router.RouteToNode(owner.ID)
+	assert.NoError(t, err)
+	assert.NotNil(t, resolvedClient)
 }
 
 func TestRouter_CircuitBreaker(t *testing.T) {