From a8af99c468c68ada8f0d87c8fffa81f774c7f921 Mon Sep 17 00:00:00 2001 From: Bolek Kulbabinski <1416262+bolekk@users.noreply.github.com> Date: Wed, 14 Oct 2026 10:28:33 -0400 Subject: [PATCH 1/5] =?UTF-8?q?cre-6798:=20migrate=20ConfidentialRelay=20g?= =?UTF-8?q?ateway=20handler=20to=20full=20DON=C3=97shard=20support?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the legacy single-shard DONConfig/DON wiring with the shared ShardedDONs/ShardEndpoint model (as Dummy/WebAPI/HTTP already use). Quorum (F+1/2F+1) and rate limiters are now evaluated per shard independently; the first shard to reach quorum wins, and a shard whose quorum becomes unreachable no longer fails the whole request on its own. Co-Authored-By: Claude Sonnet 5 --- core/services/gateway/handler_factory.go | 7 +- .../handlers/confidentialrelay/handler.go | 428 ++++++++++-------- .../confidentialrelay/handler_test.go | 218 ++++++++- 3 files changed, 459 insertions(+), 194 deletions(-) diff --git a/core/services/gateway/handler_factory.go b/core/services/gateway/handler_factory.go index 0034efa691a..05659746e46 100644 --- a/core/services/gateway/handler_factory.go +++ b/core/services/gateway/handler_factory.go @@ -84,12 +84,7 @@ func (hf *handlerFactory) NewHandler( don := dons.ConnMgrs[0][0] return vault.NewHandler(handlerConfig, donConfig, don, hf.capabilitiesRegistry, hf.workflowRegistrySyncer, hf.lggr, clockwork.NewRealClock(), hf.lf) case ConfidentialRelayHandlerType: - // For backward compatibility, convert sharded config to legacy DONConfig - // using the first DON's first shard. TODO(CRE-1640): migrate to full - // shardedDONs/shardsConnMgrs support. - donConfig := shardedDONsToLegacy(dons.DONs[0]) - don := dons.ConnMgrs[0][0] - return confidentialrelay.NewHandler(handlerConfig, donConfig, don, hf.lggr, clockwork.NewRealClock(), hf.lf) + return confidentialrelay.NewHandler(handlerConfig, dons, hf.lggr, clockwork.NewRealClock(), hf.lf) default: return nil, fmt.Errorf("unsupported handler type %s", handlerType) } diff --git a/core/services/gateway/handlers/confidentialrelay/handler.go b/core/services/gateway/handlers/confidentialrelay/handler.go index 55a61de6f68..d722b5eab92 100644 --- a/core/services/gateway/handlers/confidentialrelay/handler.go +++ b/core/services/gateway/handlers/confidentialrelay/handler.go @@ -23,7 +23,6 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings" "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" "github.com/smartcontractkit/chainlink/v2/core/services/gateway/api" - "github.com/smartcontractkit/chainlink/v2/core/services/gateway/config" gwhandlers "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers" ) @@ -204,12 +203,16 @@ type Config struct { type handler struct { services.StateMachine - donConfig *config.DONConfig - don gwhandlers.DON - codec api.JSONRPCCodec - lggr logger.Logger - mu sync.RWMutex - stopCh services.StopChan + // shards is the DON×shard matrix this handler relays for. Quorum is evaluated + // independently per shard (each with its own F and members); the first shard to + // reach quorum produces the user response. + shards []*gwhandlers.ShardEndpoint + // donID names the handler for logging/metrics; it is the first shard's DON name. + donID string + codec api.JSONRPCCodec + lggr logger.Logger + mu sync.RWMutex + stopCh services.StopChan globalNodeRateLimiter limits.RateLimiter perNodeRateLimiters map[string]limits.RateLimiter @@ -233,7 +236,7 @@ func (h *handler) Name() string { return h.lggr.Name() } -func NewHandler(methodConfig json.RawMessage, donConfig *config.DONConfig, don gwhandlers.DON, lggr logger.Logger, clock clockwork.Clock, limitsFactory limits.Factory) (*handler, error) { +func NewHandler(methodConfig json.RawMessage, dons *gwhandlers.ShardedDONs, lggr logger.Logger, clock clockwork.Clock, limitsFactory limits.Factory) (*handler, error) { var cfg Config if err := json.Unmarshal(methodConfig, &cfg); err != nil { return nil, fmt.Errorf("failed to unmarshal method config: %w", err) @@ -260,13 +263,23 @@ func NewHandler(methodConfig json.RawMessage, donConfig *config.DONConfig, don g cfg.QuorumGraceMillis = maxGraceMillis } + shards, _, err := dons.BuildShardEndpoints() + if err != nil { + return nil, err + } + if len(shards) == 0 { + return nil, errors.New("confidential relay handler requires at least one shard") + } + donID := shards[0].DonName + globalNodeRateLimiter, err := limitsFactory.MakeRateLimiter(cresettings.Default.GatewayConfidentialRelayGlobalRate) if err != nil { return nil, fmt.Errorf("failed to create global node rate limiter: %w", err) } - perNodeRateLimiters := make(map[string]limits.RateLimiter, len(donConfig.Members)) - for _, member := range donConfig.Members { + members := gwhandlers.AllMembers(shards) + perNodeRateLimiters := make(map[string]limits.RateLimiter, len(members)) + for _, member := range members { rl, makeErr := limitsFactory.MakeRateLimiter(cresettings.Default.GatewayConfidentialRelayPerNodeRate) if makeErr != nil { return nil, fmt.Errorf("failed to create per-node rate limiter for %s: %w", member.Address, makeErr) @@ -280,9 +293,9 @@ func NewHandler(methodConfig json.RawMessage, donConfig *config.DONConfig, don g } return &handler{ - donConfig: donConfig, - don: don, - lggr: logger.Named(lggr, "ConfidentialRelayHandler:"+donConfig.DonID), + shards: shards, + donID: donID, + lggr: logger.Named(lggr, "ConfidentialRelayHandler:"+donID), requestTimeout: time.Duration(cfg.RequestTimeoutSec) * time.Second, nodeSendTimeout: time.Duration(cfg.NodeSendTimeoutSec) * time.Second, quorumGrace: time.Duration(cfg.QuorumGraceMillis) * time.Millisecond, @@ -297,6 +310,15 @@ func NewHandler(methodConfig json.RawMessage, donConfig *config.DONConfig, don g }, nil } +// totalNodes is the number of relay nodes across every shard, used only for logging. +func (h *handler) totalNodes() int { + n := 0 + for _, shard := range h.shards { + n += len(shard.Members) + } + return n +} + func (h *handler) Start(_ context.Context) error { return h.StartOnce("ConfidentialRelayHandler", func() error { h.lggr.Info("starting confidential relay handler") @@ -346,25 +368,18 @@ func (h *handler) removeExpiredRequests(ctx context.Context) { h.mu.RUnlock() for _, er := range expiredRequests { - responses := er.copiedResponses() l := h.requestLogger(er.req, er.labels) + collected := len(er.copiedResponses()) l.Debugw("request expired, evaluating collected relay responses", - "collected", len(responses), - "nodes", len(h.donConfig.Members), - "unanswered", len(h.donConfig.Members)-len(responses), + "collected", collected, + "nodes", h.totalNodes(), + "unanswered", h.totalNodes()-collected, ) - summary, err := h.bundler.Bundle(er.req, responses, l) - if err != nil { - l.Errorw("failed to build relay response bundle", "error", err) - if sendErr := h.sendResponseAndClearRequest(ctx, er, h.constructErrorResponse(er.req, api.FatalError, err)); sendErr != nil { - l.Errorw("error returning bundle failure on expiry", "error", sendErr) - } - continue - } // Expiry makes further responses unavailable to this request. The common - // readiness path forwards a viable partial bundle or returns a timeout. - if err := h.forwardBundleOrTerminateIfReady(ctx, l, er, summary, 0, true); err != nil { - l.Errorw("error forwarding bundle on expiry", "error", err) + // readiness path forwards a viable partial bundle from whichever shard + // qualifies, or returns a timeout. + if err := h.evaluateShards(ctx, l, er, true); err != nil { + l.Errorw("error evaluating relay response bundles on expiry", "error", err) } } } @@ -401,7 +416,7 @@ func (h *handler) HandleJSONRPCUserMessage(ctx context.Context, req jsonrpc.Requ labels := h.extractRequestLabels(req) l := h.requestLogger(req, labels) - l.Debugw("handling confidential relay request", "nodes", len(h.donConfig.Members), "f", h.donConfig.F) + l.Debugw("handling confidential relay request", "shards", len(h.shards), "nodes", h.totalNodes()) ar, err := h.newActiveRequest(req, labels, callback) if err != nil { @@ -467,122 +482,152 @@ func (h *handler) HandleNodeMessage(ctx context.Context, resp *jsonrpc.Response[ return nil } - copiedResponses := ar.copiedResponses() - // Bundle readiness counts decodable signed responses, while the enclave remains - // responsible for signature and quorum verification. A collected response cannot - // be replaced by a later response from the same node. - summary, err := h.bundler.Bundle(ar.req, copiedResponses, l) - if err != nil { - l.Errorw("failed to build relay response bundle", "error", err) - return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.FatalError, err)) - } - return h.forwardBundleOrTerminateIfReady(ctx, l, ar, summary, len(h.donConfig.Members)-summary.Total(), false) -} - -// forwardBundleOrTerminateIfReady applies the same completion rules to live and -// expired requests. Remaining counts responses that can still arrive; expiry -// passes zero because waiting can no longer improve the bundle. -func (h *handler) forwardBundleOrTerminateIfReady(ctx context.Context, l logger.Logger, ar *activeRequest, summary *BundleSummary, remaining int, expired bool) error { - earlyNeed := 2*h.donConfig.F + 1 - minQuorum := h.donConfig.F + 1 - nodes := len(h.donConfig.Members) - unanswered := nodes - summary.Total() - maxPossibleSigned := summary.Signed() + remaining - if summary.Signed() >= earlyNeed { - return h.forwardBundle(ctx, l, ar, summary) - } - if remaining == 0 && summary.Signed() >= minQuorum { - l.Infow("no more responses expected; forwarding partial signed bundle", - "signed", summary.Signed(), - "minQuorum", minQuorum, - "earlyNeed", earlyNeed, - "collected", summary.Total(), - "nodes", nodes, - "unanswered", unanswered, - "expired", expired, - "errors", summary.Error(), - "undecodable", summary.Undecodable(), - "nodeErrors", summary.NodeErrorsFormatted(), - ) - return h.forwardBundle(ctx, l, ar, summary) + return h.evaluateShards(ctx, l, ar, false) +} + +// filterResponses restricts a request's collected responses to one shard's members. +// Node addresses are disjoint across shards (enforced by BuildShardEndpoints), so this +// never pulls in a response from another shard. +func filterResponses(all map[string]jsonrpc.Response[json.RawMessage], shard *gwhandlers.ShardEndpoint) map[string]jsonrpc.Response[json.RawMessage] { + out := make(map[string]jsonrpc.Response[json.RawMessage], len(shard.Members)) + for _, m := range shard.Members { + if r, ok := all[m.Address]; ok { + out[m.Address] = r + } } - if maxPossibleSigned < minQuorum { - // A user-level node error means the request is invalid for every node, - // so report it rather than the generic quorum failure. - if userErr := summary.UserError(); userErr != nil { - l.Warnw("relay quorum unreachable due to a user error; propagating node error", - "signed", summary.Signed(), - "minQuorum", minQuorum, - "collected", summary.Total(), - "nodes", nodes, - "remaining", remaining, - "expired", expired, - "errors", summary.Error(), - "undecodable", summary.Undecodable(), - "userErrorCode", userErr.Code, - "nodeErrors", summary.NodeErrorsFormatted(), - ) - return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.InvalidParamsError, - errors.New(sanitizeNodeErrorMessage(userErr.Message)))) + return out +} + +// evaluateShards applies the completion rules to live and expired requests, once per +// shard: each shard's quorum is evaluated against its own F and members, independently +// of every other shard, and the first shard whose bundle is ready wins the request. +// Only when every shard can no longer reach its own quorum does the request fail. +// Expiry makes further responses unavailable, so remaining is forced to zero for every +// shard rather than counted. +func (h *handler) evaluateShards(ctx context.Context, l logger.Logger, ar *activeRequest, expired bool) error { + all := ar.copiedResponses() + + var ( + allDead = true + firstUserErr *jsonrpc.WireError + totalSigned int + totalCollected int + ) + + for _, shard := range h.shards { + summary, err := h.bundler.Bundle(ar.req, filterResponses(all, shard), l) + if err != nil { + l.Errorw("failed to build relay response bundle", "shard", shard.DonID, "error", err) + return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.FatalError, err)) } + + minQuorum := shard.F + 1 + earlyNeed := 2*shard.F + 1 + nodes := len(shard.Members) + remaining := nodes - summary.Total() if expired { - l.Warnw("request expired before relay quorum was reached", + remaining = 0 + } + maxPossibleSigned := summary.Signed() + remaining + + totalSigned += summary.Signed() + totalCollected += summary.Total() + + if summary.Signed() >= earlyNeed { + return h.forwardBundle(ctx, l, ar, shard, summary) + } + if remaining == 0 && summary.Signed() >= minQuorum { + l.Infow("no more responses expected in shard; forwarding partial signed bundle", + "shard", shard.DonID, "signed", summary.Signed(), "minQuorum", minQuorum, "earlyNeed", earlyNeed, "collected", summary.Total(), "nodes", nodes, - "unanswered", unanswered, + "unanswered", nodes-summary.Total(), + "expired", expired, "errors", summary.Error(), "undecodable", summary.Undecodable(), "nodeErrors", summary.NodeErrorsFormatted(), ) - return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.RequestTimeoutError, - fmt.Errorf("request expired: %d signed relay responses, need at least %d to reach quorum (collected=%d nodes=%d unanswered=%d errors=%d undecodable=%d)", - summary.Signed(), minQuorum, summary.Total(), nodes, unanswered, summary.Error(), summary.Undecodable()))) + return h.forwardBundle(ctx, l, ar, shard, summary) } - l.Warnw("relay quorum unreachable; returning error without waiting", - "signed", summary.Signed(), - "minQuorum", minQuorum, - "earlyNeed", earlyNeed, - "collected", summary.Total(), + if maxPossibleSigned < minQuorum { + // A user-level node error means the request is invalid for every node, + // so report it rather than the generic quorum failure. + if userErr := summary.UserError(); userErr != nil && firstUserErr == nil { + firstUserErr = userErr + } + continue + } + + allDead = false + if !expired && summary.Signed() >= minQuorum { + h.startQuorumGrace(l, ar, summary, shard) + } + } + + if allDead { + return h.failUnreachableQuorum(ctx, l, ar, firstUserErr, expired, totalSigned, totalCollected) + } + + l.Debugw("waiting for more signed relay responses before forwarding bundle", + "signed", totalSigned, + "collected", totalCollected, + "nodes", h.totalNodes(), + "shards", len(h.shards), + ) + return nil +} + +// failUnreachableQuorum is reached only once every shard's quorum has become +// unreachable: a single-shard "quorum unreachable" verdict no longer fails the +// request outright, since another shard may still complete it. +func (h *handler) failUnreachableQuorum(ctx context.Context, l logger.Logger, ar *activeRequest, userErr *jsonrpc.WireError, expired bool, signed, collected int) error { + nodes := h.totalNodes() + if userErr != nil { + l.Warnw("relay quorum unreachable in every shard due to a user error; propagating node error", + "signed", signed, + "collected", collected, "nodes", nodes, - "remaining", remaining, - "maxPossibleSigned", maxPossibleSigned, - "errors", summary.Error(), - "undecodable", summary.Undecodable(), - "nodeErrors", summary.NodeErrorsFormatted(), + "shards", len(h.shards), + "expired", expired, + "userErrorCode", userErr.Code, ) - return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.FatalError, - fmt.Errorf("relay quorum unreachable: %d signed responses, at most %d possible, need %d (collected=%d nodes=%d remaining=%d errors=%d undecodable=%d)", - summary.Signed(), maxPossibleSigned, minQuorum, summary.Total(), nodes, remaining, summary.Error(), summary.Undecodable()))) + return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.InvalidParamsError, + errors.New(sanitizeNodeErrorMessage(userErr.Message)))) } - if !expired && summary.Signed() >= minQuorum { - h.startQuorumGrace(l, ar, summary) + if expired { + l.Warnw("request expired before relay quorum was reached in any shard", + "signed", signed, + "collected", collected, + "nodes", nodes, + "shards", len(h.shards), + ) + return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.RequestTimeoutError, + fmt.Errorf("request expired: %d signed relay responses across %d shard(s), none reached quorum (collected=%d nodes=%d)", + signed, len(h.shards), collected, nodes))) } - l.Debugw("waiting for more signed relay responses before forwarding bundle", - "signed", summary.Signed(), - "earlyNeed", earlyNeed, - "minQuorum", minQuorum, - "collected", summary.Total(), + l.Warnw("relay quorum unreachable in every shard; returning error without waiting", + "signed", signed, + "collected", collected, "nodes", nodes, - "remaining", remaining, - "maxPossibleSigned", maxPossibleSigned, - "errors", summary.Error(), - "undecodable", summary.Undecodable(), - "nodeErrors", summary.NodeErrorsFormatted(), + "shards", len(h.shards), ) - return nil + return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.FatalError, + fmt.Errorf("relay quorum unreachable: %d signed responses across %d shard(s), none can reach quorum (collected=%d nodes=%d)", + signed, len(h.shards), collected, nodes))) } // startQuorumGrace arms the grace window the first time a request holds minQuorum -// signed responses. It bounds how long a request that will never reach earlyNeed -// keeps waiting for the rest of the DON: without it such a request is held until -// requestTimeout, long after the caller's own deadline has elapsed, so a bundle +// signed responses in some shard. It bounds how long a request that will never reach +// earlyNeed keeps waiting for the rest of the DON: without it such a request is held +// until requestTimeout, long after the caller's own deadline has elapsed, so a bundle // that was viable within milliseconds is forwarded to nobody. The deadline is swept // by the cleanup goroutine rather than a timer, so no completion path runs on an -// untracked goroutine. -func (h *handler) startQuorumGrace(l logger.Logger, ar *activeRequest, summary *BundleSummary) { +// untracked goroutine. It is armed once per request (not per shard): the first shard +// to reach minQuorum starts the clock for whichever shard ends up winning. +func (h *handler) startQuorumGrace(l logger.Logger, ar *activeRequest, summary *BundleSummary, shard *gwhandlers.ShardEndpoint) { if h.quorumGrace <= 0 { return } @@ -590,11 +635,12 @@ func (h *handler) startQuorumGrace(l logger.Logger, ar *activeRequest, summary * return } l.Infow("relay quorum reached below earlyNeed; starting grace window", + "shard", shard.DonID, "signed", summary.Signed(), - "minQuorum", h.donConfig.F+1, - "earlyNeed", 2*h.donConfig.F+1, + "minQuorum", shard.F+1, + "earlyNeed", 2*shard.F+1, "collected", summary.Total(), - "nodes", len(h.donConfig.Members), + "nodes", len(shard.Members), "grace", h.quorumGrace, ) } @@ -620,56 +666,59 @@ func (h *handler) forwardGracedRequests(ctx context.Context) { func (h *handler) forwardAfterGrace(ctx context.Context, ar *activeRequest) { l := h.requestLogger(ar.req, ar.labels) - summary, err := h.bundler.Bundle(ar.req, ar.copiedResponses(), l) - if err != nil { - l.Errorw("failed to build relay response bundle after quorum grace", "error", err) - if sendErr := h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.FatalError, err)); sendErr != nil { - l.Errorw("error returning bundle failure after quorum grace", "error", sendErr) + all := ar.copiedResponses() + + for _, shard := range h.shards { + summary, err := h.bundler.Bundle(ar.req, filterResponses(all, shard), l) + if err != nil { + l.Errorw("failed to build relay response bundle after quorum grace", "shard", shard.DonID, "error", err) + if sendErr := h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.FatalError, err)); sendErr != nil { + l.Errorw("error returning bundle failure after quorum grace", "error", sendErr) + } + return } - return - } - minQuorum := h.donConfig.F + 1 - nodes := len(h.donConfig.Members) - if summary.Signed() < minQuorum { - l.Warnw("quorum grace elapsed below quorum; leaving request to expiry", + minQuorum := shard.F + 1 + if summary.Signed() < minQuorum { + continue + } + + nodes := len(shard.Members) + l.Infow("quorum grace elapsed; forwarding partial signed bundle", + "shard", shard.DonID, "signed", summary.Signed(), "minQuorum", minQuorum, + "earlyNeed", 2*shard.F+1, "collected", summary.Total(), "nodes", nodes, + "unanswered", nodes-summary.Total(), + "grace", h.quorumGrace, + "errors", summary.Error(), + "undecodable", summary.Undecodable(), + "nodeErrors", summary.NodeErrorsFormatted(), ) + if err := h.forwardBundle(ctx, l, ar, shard, summary); err != nil { + l.Errorw("error forwarding bundle after quorum grace", "error", err) + } return } - l.Infow("quorum grace elapsed; forwarding partial signed bundle", - "signed", summary.Signed(), - "minQuorum", minQuorum, - "earlyNeed", 2*h.donConfig.F+1, - "collected", summary.Total(), - "nodes", nodes, - "unanswered", nodes-summary.Total(), - "grace", h.quorumGrace, - "errors", summary.Error(), - "undecodable", summary.Undecodable(), - "nodeErrors", summary.NodeErrorsFormatted(), - ) - if err := h.forwardBundle(ctx, l, ar, summary); err != nil { - l.Errorw("error forwarding bundle after quorum grace", "error", err) - } + l.Warnw("quorum grace elapsed without any shard reaching quorum; leaving request to expiry") } -// forwardBundle sends a previously-built bundle to the enclave. The gateway makes +// forwardBundle sends a previously-built shard bundle to the enclave. The gateway makes // no trust decision; the enclave verifies signatures and reaches quorum. -func (h *handler) forwardBundle(ctx context.Context, l logger.Logger, ar *activeRequest, summary *BundleSummary) error { - nodes := len(h.donConfig.Members) +func (h *handler) forwardBundle(ctx context.Context, l logger.Logger, ar *activeRequest, shard *gwhandlers.ShardEndpoint, summary *BundleSummary) error { + nodes := len(shard.Members) rawResponse, err := jsonrpc.EncodeResponse(summary.Response()) if err != nil { l.Errorw("failed to encode response", "error", err) return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.NodeReponseEncodingError, err)) } l.Infow("forwarding relay response bundle to enclave", + "shard", shard.DonID, "signedResponses", summary.Signed(), - "earlyNeed", 2*h.donConfig.F+1, + "earlyNeed", 2*shard.F+1, "collected", summary.Total(), "nodes", nodes, "remaining", nodes-summary.Total(), @@ -683,11 +732,13 @@ func (h *handler) forwardBundle(ctx context.Context, l logger.Logger, ar *active }) } +// fanOutToNodes sends the request to every member of every shard, each via that +// shard's own connection manager. A shard whose own quorum becomes unreachable here +// does not fail the request by itself: another shard may still complete it. The +// request only fails if every shard is left unreachable. func (h *handler) fanOutToNodes(ctx context.Context, l logger.Logger, ar *activeRequest) error { - var ( - group errgroup.Group - nodeErrors atomic.Uint32 - ) + var group errgroup.Group + shardNodeErrors := make([]atomic.Uint32, len(h.shards)) // Each send is bounded independently. A node whose websocket accepts no writes blocks // until its context is cancelled, and because the caller only reads the response callback @@ -696,34 +747,47 @@ func (h *handler) fanOutToNodes(ctx context.Context, l logger.Logger, ar *active sendCtx, cancel := context.WithTimeout(ctx, h.nodeSendTimeout) defer cancel() - for _, node := range h.donConfig.Members { - group.Go(func() error { - err := h.don.SendToNode(sendCtx, node.Address, &ar.req) - if err != nil { - nodeErrors.Add(1) - l.Errorw("error sending request to node", "node", node.Address, "error", err) - } - return nil - }) + for i, shard := range h.shards { + for _, node := range shard.Members { + group.Go(func() error { + err := shard.ConnMgr.SendToNode(sendCtx, node.Address, &ar.req) + if err != nil { + shardNodeErrors[i].Add(1) + l.Errorw("error sending request to node", "node", node.Address, "shard", shard.DonID, "error", err) + } + return nil + }) + } } _ = group.Wait() - numNodeErrors := nodeErrors.Load() - remainingPossibleResponses := len(h.donConfig.Members) - int(numNodeErrors) - if remainingPossibleResponses < h.donConfig.F+1 && numNodeErrors > 0 { - // The individual send failures are logged above; this is the aggregate - // decision that ends the request, so it gets its own line. - l.Errorw("too few relay nodes reachable to reach quorum; failing request", - "nodeErrors", numNodeErrors, - "reachable", remainingPossibleResponses, - "minQuorum", h.donConfig.F+1, - "nodes", len(h.donConfig.Members), - ) + var anyShardReachable bool + var totalNodeErrors uint32 + for i, shard := range h.shards { + numNodeErrors := shardNodeErrors[i].Load() + totalNodeErrors += numNodeErrors + reachable := len(shard.Members) - int(numNodeErrors) + if reachable >= shard.F+1 { + anyShardReachable = true + } else if numNodeErrors > 0 { + l.Errorw("too few relay nodes reachable in shard to reach quorum", + "shard", shard.DonID, + "nodeErrors", numNodeErrors, + "reachable", reachable, + "minQuorum", shard.F+1, + "nodes", len(shard.Members), + ) + } + } + if !anyShardReachable { + // The per-shard failures are logged above; this is the aggregate decision + // that ends the request, so it gets its own line. + l.Errorw("too few relay nodes reachable in any shard to reach quorum; failing request", "nodeErrors", totalNodeErrors) return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.FatalError, errors.New("failed to forward user request to nodes"))) } - l.Debugw("successfully forwarded request to relay nodes", "nodeErrors", numNodeErrors) + l.Debugw("successfully forwarded request to relay nodes", "nodeErrors", totalNodeErrors) return nil } @@ -759,16 +823,16 @@ func (h *handler) recordMetrics(ctx context.Context, errorCode api.ErrorCode) { switch errorCode { case api.HandlerError: h.metrics.requestInternalError.Add(ctx, 1, metric.WithAttributes( - attribute.String("don_id", h.donConfig.DonID), + attribute.String("don_id", h.donID), attribute.String("error", errorCode.String()), )) case api.UnsupportedDONIdError: h.metrics.requestUserError.Add(ctx, 1, metric.WithAttributes( - attribute.String("don_id", h.donConfig.DonID), + attribute.String("don_id", h.donID), )) case api.NoError: h.metrics.requestSuccess.Add(ctx, 1, metric.WithAttributes( - attribute.String("don_id", h.donConfig.DonID), + attribute.String("don_id", h.donID), )) } } diff --git a/core/services/gateway/handlers/confidentialrelay/handler_test.go b/core/services/gateway/handlers/confidentialrelay/handler_test.go index b84942daf29..3b82bfb9ca8 100644 --- a/core/services/gateway/handlers/confidentialrelay/handler_test.go +++ b/core/services/gateway/handlers/confidentialrelay/handler_test.go @@ -23,10 +23,28 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" "github.com/smartcontractkit/chainlink/v2/core/services/gateway/api" "github.com/smartcontractkit/chainlink/v2/core/services/gateway/config" + gwhandlers "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers" "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/common" "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/mocks" ) +// singleShardDONs wraps one legacy single-shard DONConfig/DON pair into the +// *gwhandlers.ShardedDONs matrix NewHandler now takes, so existing single-shard test +// setups keep their exact F/Members/node-address expectations. +func singleShardDONs(t *testing.T, donConfig *config.DONConfig, don gwhandlers.DON) *gwhandlers.ShardedDONs { + t.Helper() + dons, err := gwhandlers.NewShardedDONs( + []config.ShardedDONConfig{{ + DonName: donConfig.DonID, + F: donConfig.F, + Shards: []config.Shard{{Nodes: donConfig.Members}}, + }}, + [][]gwhandlers.DON{{don}}, + ) + require.NoError(t, err) + return dons +} + type barrierDON struct { total int mu sync.Mutex @@ -133,7 +151,7 @@ func setupHandlerWithConfig(t *testing.T, numNodes, f int, handlerConfig Config) clock := clockwork.NewFakeClock() limitsFactory := limits.Factory{Settings: cresettings.DefaultGetter, Logger: lggr} - h, err := NewHandler(methodConfig, donConfig, don, lggr, clock, limitsFactory) + h, err := NewHandler(methodConfig, singleShardDONs(t, donConfig, don), lggr, clock, limitsFactory) require.NoError(t, err) // The default bundler is the real one; tests that need a failure inject a mock. cb := common.NewCallback() @@ -851,7 +869,7 @@ func TestConfidentialRelayHandler_QuorumGraceConfig(t *testing.T) { donConfig := &config.DONConfig{DonID: "test_relay_don", F: 1, Members: []config.NodeConfig{nodeOne}} limitsFactory := limits.Factory{Settings: cresettings.DefaultGetter, Logger: lggr} - h, err := NewHandler(methodConfig, donConfig, mocks.NewDON(t), lggr, clockwork.NewFakeClock(), limitsFactory) + h, err := NewHandler(methodConfig, singleShardDONs(t, donConfig, mocks.NewDON(t)), lggr, clockwork.NewFakeClock(), limitsFactory) require.NoError(t, err) assert.Equal(t, tc.wantGraceMs, h.quorumGrace.Milliseconds()) @@ -899,7 +917,7 @@ func TestConfidentialRelayHandler_RateLimitedNode(t *testing.T) { } clock := clockwork.NewFakeClock() limitsFactory := limits.Factory{Settings: cresettings.DefaultGetter, Logger: lggr} - h, err := NewHandler(methodConfig, donConfig, don, lggr, clock, limitsFactory) + h, err := NewHandler(methodConfig, singleShardDONs(t, donConfig, don), lggr, clock, limitsFactory) require.NoError(t, err) h.globalNodeRateLimiter = limits.GlobalRateLimiter(rate.Limit(100), 100) h.perNodeRateLimiters[nodeOne.Address] = limits.GlobalRateLimiter(rate.Limit(0.001), 1) @@ -1113,7 +1131,7 @@ func TestConfidentialRelayHandler_FanOutToNodes_IsConcurrent(t *testing.T) { require.NoError(t, err) limitsFactory := limits.Factory{Settings: cresettings.DefaultGetter, Logger: lggr} - h, err := NewHandler(methodConfig, donConfig, don, lggr, clockwork.NewFakeClock(), limitsFactory) + h, err := NewHandler(methodConfig, singleShardDONs(t, donConfig, don), lggr, clockwork.NewFakeClock(), limitsFactory) require.NoError(t, err) cb := common.NewCallback() @@ -1166,7 +1184,7 @@ func TestConfidentialRelayHandler_NodeSendTimeoutConfig(t *testing.T) { donConfig := &config.DONConfig{DonID: "test_relay_don", F: 1, Members: []config.NodeConfig{nodeOne}} limitsFactory := limits.Factory{Settings: cresettings.DefaultGetter, Logger: lggr} - h, err := NewHandler(methodConfig, donConfig, mocks.NewDON(t), lggr, clockwork.NewFakeClock(), limitsFactory) + h, err := NewHandler(methodConfig, singleShardDONs(t, donConfig, mocks.NewDON(t)), lggr, clockwork.NewFakeClock(), limitsFactory) require.NoError(t, err) assert.Equal(t, tc.wantSendMs, h.nodeSendTimeout.Milliseconds()) @@ -1197,7 +1215,7 @@ func TestConfidentialRelayHandler_BlockedNodeDoesNotStallFanOut(t *testing.T) { methodConfig, err := json.Marshal(Config{RequestTimeoutSec: 30}) require.NoError(t, err) limitsFactory := limits.Factory{Settings: cresettings.DefaultGetter, Logger: lggr} - h, err := NewHandler(methodConfig, donConfig, don, lggr, clockwork.NewFakeClock(), limitsFactory) + h, err := NewHandler(methodConfig, singleShardDONs(t, donConfig, don), lggr, clockwork.NewFakeClock(), limitsFactory) require.NoError(t, err) // Shortened so the test exercises the bound without waiting the production default. h.nodeSendTimeout = 50 * time.Millisecond @@ -1232,3 +1250,191 @@ func capExecSignedRespPtr(t *testing.T, id string, result relaytypes.CapabilityR r := capExecSignedResponse(t, id, result, signer) return &r } + +// setupMultiShardHandler builds a handler backed by one DON split into shards, one per +// entry in shardSizes, every shard sharing F. Node addresses are assigned sequentially +// across shards (shard 0 gets the first shardSizes[0] addresses, and so on), and each +// shard gets its own mock connection manager so tests can assert routing stays confined +// to the owning shard. +func setupMultiShardHandler(t *testing.T, f int, shardSizes ...int) (*handler, *common.Callback, []*mocks.DON, *clockwork.FakeClock) { + t.Helper() + lggr := logger.Test(t) + + shards := make([]config.Shard, len(shardSizes)) + conns := make([]gwhandlers.DON, len(shardSizes)) + mockConns := make([]*mocks.DON, len(shardSizes)) + addr := 0 + for i, size := range shardSizes { + members := make([]config.NodeConfig, size) + for j := range size { + members[j] = config.NodeConfig{Name: fmt.Sprintf("node%d", addr), Address: fmt.Sprintf("0x%04d", addr)} + addr++ + } + shards[i] = config.Shard{Nodes: members} + don := mocks.NewDON(t) + conns[i] = don + mockConns[i] = don + } + + dons, err := gwhandlers.NewShardedDONs( + []config.ShardedDONConfig{{DonName: "test_relay_don", F: f, Shards: shards}}, + [][]gwhandlers.DON{conns}, + ) + require.NoError(t, err) + + methodConfig, err := json.Marshal(Config{RequestTimeoutSec: 30}) + require.NoError(t, err) + + clock := clockwork.NewFakeClock() + limitsFactory := limits.Factory{Settings: cresettings.DefaultGetter, Logger: lggr} + h, err := NewHandler(methodConfig, dons, lggr, clock, limitsFactory) + require.NoError(t, err) + cb := common.NewCallback() + return h, cb, mockConns, clock +} + +func nodeErrorResponse(id string) *jsonrpc.Response[json.RawMessage] { + return &jsonrpc.Response[json.RawMessage]{ + Version: jsonrpc.JsonRpcVersion, + ID: id, + Method: MethodCapabilityExec, + Error: &jsonrpc.WireError{Code: -32603, Message: "internal error"}, + } +} + +// First shard to reach quorum wins: shard 0 reaches 2F+1 while shard 1 has no +// responses at all, and the forwarded bundle contains only shard 0's responses. +func TestConfidentialRelayHandler_Sharded_FirstShardToReachQuorumWins(t *testing.T) { + t.Parallel() + h, cb, conns, _ := setupMultiShardHandler(t, 1, 4, 4) // F=1: minQuorum=2, earlyNeed=3 per shard + for _, conn := range conns { + conn.On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(nil) + } + + params := validCapParamsJSON("wf1") + req := jsonrpc.Request[json.RawMessage]{ + ID: "req-sharded-win", + Method: MethodCapabilityExec, + Params: ¶ms, + } + result := relaytypes.CapabilityResponseResult{Payload: "result"} + + require.NoError(t, h.HandleJSONRPCUserMessage(t.Context(), req, cb)) + + var wg sync.WaitGroup + wg.Go(func() { + resp, err := cb.Wait(t.Context()) + if !assert.NoError(t, err) { //nolint:testifylint // require illegal inside wg.Go goroutine + return + } + assert.Equal(t, api.NoError, resp.ErrorCode) + var jsonResp jsonrpc.Response[json.RawMessage] + if !assert.NoError(t, json.Unmarshal(resp.RawResponse, &jsonResp)) { + return + } + var bundle relaytypes.SignedCapabilityResponseBundle + if !assert.NoError(t, json.Unmarshal(*jsonResp.Result, &bundle)) { + return + } + assert.Len(t, bundle.Responses, 3, "only shard 0's signed responses are bundled") + }) + + // Shard 0 is nodes 0x0000-0x0003; reaching 3 signed responses hits 2F+1 there. + for i := range 3 { + require.NoError(t, h.HandleNodeMessage(t.Context(), + capExecSignedRespPtr(t, req.ID, result, fmt.Appendf(nil, "signer-%d", i)), + fmt.Sprintf("0x%04d", i))) + } + wg.Wait() + require.Nil(t, h.getActiveRequest(req.ID), "request completes once any shard reaches quorum") + + // A late response from shard 1 (nodes 0x0004-0x0007) must be ignored: the request + // is already answered, and shard 1's quorum is irrelevant once shard 0 has won. + require.NoError(t, h.HandleNodeMessage(t.Context(), + capExecSignedRespPtr(t, req.ID, result, []byte("signer-late")), "0x0004")) +} + +// A shard whose own quorum becomes unreachable does not fail the request: another +// shard can still complete it independently. +func TestConfidentialRelayHandler_Sharded_DeadShardDoesNotFailRequest(t *testing.T) { + t.Parallel() + h, cb, conns, _ := setupMultiShardHandler(t, 1, 4, 4) + for _, conn := range conns { + conn.On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(nil) + } + + params := validCapParamsJSON("wf1") + req := jsonrpc.Request[json.RawMessage]{ + ID: "req-sharded-dead", + Method: MethodCapabilityExec, + Params: ¶ms, + } + result := relaytypes.CapabilityResponseResult{Payload: "result"} + + require.NoError(t, h.HandleJSONRPCUserMessage(t.Context(), req, cb)) + + // Shard 0 (nodes 0x0000-0x0003) answers entirely with errors: once every member + // has replied, signed stays 0 and this shard can never reach minQuorum=2. + for i := range 4 { + require.NoError(t, h.HandleNodeMessage(t.Context(), nodeErrorResponse(req.ID), fmt.Sprintf("0x%04d", i))) + } + require.NotNil(t, h.getActiveRequest(req.ID), "shard 1 can still reach quorum") + + var wg sync.WaitGroup + wg.Go(func() { + resp, err := cb.Wait(t.Context()) + if !assert.NoError(t, err) { //nolint:testifylint // require illegal inside wg.Go goroutine + return + } + assert.Equal(t, api.NoError, resp.ErrorCode) + }) + + // Shard 1 (nodes 0x0004-0x0007) reaches 2F+1=3 signed responses. + for i := 4; i < 7; i++ { + require.NoError(t, h.HandleNodeMessage(t.Context(), + capExecSignedRespPtr(t, req.ID, result, fmt.Appendf(nil, "signer-%d", i)), + fmt.Sprintf("0x%04d", i))) + } + wg.Wait() + require.Nil(t, h.getActiveRequest(req.ID)) +} + +// Only once every shard's quorum is unreachable does the request fail. +func TestConfidentialRelayHandler_Sharded_AllShardsDeadFailsRequest(t *testing.T) { + t.Parallel() + h, cb, conns, _ := setupMultiShardHandler(t, 1, 4, 4) + for _, conn := range conns { + conn.On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(nil) + } + + params := validCapParamsJSON("wf1") + req := jsonrpc.Request[json.RawMessage]{ + ID: "req-sharded-all-dead", + Method: MethodCapabilityExec, + Params: ¶ms, + } + + var wg sync.WaitGroup + wg.Go(func() { + resp, err := cb.Wait(t.Context()) + if !assert.NoError(t, err) { //nolint:testifylint // require illegal inside wg.Go goroutine + return + } + assert.Equal(t, api.FatalError, resp.ErrorCode) + var jsonResp jsonrpc.Response[json.RawMessage] + if !assert.NoError(t, json.Unmarshal(resp.RawResponse, &jsonResp)) { //nolint:testifylint // require illegal inside wg.Go goroutine + return + } + if assert.NotNil(t, jsonResp.Error) { + assert.Contains(t, jsonResp.Error.Message, "relay quorum unreachable") + } + }) + + require.NoError(t, h.HandleJSONRPCUserMessage(t.Context(), req, cb)) + + for i := range 8 { + require.NoError(t, h.HandleNodeMessage(t.Context(), nodeErrorResponse(req.ID), fmt.Sprintf("0x%04d", i))) + } + wg.Wait() + require.Nil(t, h.getActiveRequest(req.ID)) +} From c679cac435aff0a3e86359590b199f7c0f32133c Mon Sep 17 00:00:00 2001 From: Bolek Kulbabinski <1416262+bolekk@users.noreply.github.com> Date: Wed, 14 Oct 2026 17:38:45 -0400 Subject: [PATCH 2/5] cre-6798: route confidential relay requests to a single shard Fan-out previously raced every configured shard and forwarded whichever reached quorum first, which is wrong here: secrets.get/capability.exec requests are not interchangeable across shards. The enclave now selects the shard via an explicit shard_index in request params (bumped to chainlink-common's new *int ShardIndex field, PR smartcontractkit/chainlink-common#2470); a missing value is rejected rather than defaulted to shard 0, since 0 is itself a valid shard. fanOutToNodes, evaluateShard (renamed from evaluateShards), and forwardAfterGrace now operate on the single resolved shard instead of looping over all of them. Co-Authored-By: Claude Sonnet 5 --- .../confidentialrelay/bundler_test.go | 2 + .../handlers/confidentialrelay/handler.go | 400 +++++++++--------- .../confidentialrelay/handler_test.go | 142 ++----- go.mod | 2 +- go.sum | 2 + 5 files changed, 253 insertions(+), 295 deletions(-) diff --git a/core/services/gateway/handlers/confidentialrelay/bundler_test.go b/core/services/gateway/handlers/confidentialrelay/bundler_test.go index 139bbfa7350..23370c8043c 100644 --- a/core/services/gateway/handlers/confidentialrelay/bundler_test.go +++ b/core/services/gateway/handlers/confidentialrelay/bundler_test.go @@ -30,6 +30,7 @@ func validCapParams(workflowID string) relaytypes.CapabilityRequestParams { ReferenceID: "ref-1", CapabilityID: "cap-1", Payload: "in", + ShardIndex: new(0), } } @@ -40,6 +41,7 @@ func validSecretsParams(workflowID string) relaytypes.SecretsRequestParams { ExecutionID: testExecutionID, Secrets: []relaytypes.SecretIdentifier{{Key: "k1", Namespace: "ns"}}, EnclavePublicKey: testEnclavePK, + ShardIndex: new(0), } } diff --git a/core/services/gateway/handlers/confidentialrelay/handler.go b/core/services/gateway/handlers/confidentialrelay/handler.go index d722b5eab92..e8c986a9ae3 100644 --- a/core/services/gateway/handlers/confidentialrelay/handler.go +++ b/core/services/gateway/handlers/confidentialrelay/handler.go @@ -106,9 +106,42 @@ func (h *handler) extractRequestLabels(req jsonrpc.Request[json.RawMessage]) req return labels } +// requestShardSelector is the shard-routing field both relay methods' params carry. +// Unlike requestLabels, decoding this is not best-effort: a request that omits it, or +// names a shard the handler does not have, is rejected rather than silently routed to +// shard 0, since 0 is itself a valid shard and must not be indistinguishable from "not +// set." +type requestShardSelector struct { + ShardIndex *int `json:"shard_index"` +} + +// extractShardIndex decodes the caller-selected shard index out of a request's params. +func extractShardIndex(req jsonrpc.Request[json.RawMessage]) (int, error) { + if req.Params == nil { + return 0, errors.New("shard_index is required") + } + var sel requestShardSelector + if err := json.Unmarshal(*req.Params, &sel); err != nil { + return 0, fmt.Errorf("could not decode shard_index: %w", err) + } + if sel.ShardIndex == nil { + return 0, errors.New("shard_index is required") + } + return *sel.ShardIndex, nil +} + +// shardAt resolves a caller-selected shard index to the shard it names. +func (h *handler) shardAt(idx int) (*gwhandlers.ShardEndpoint, error) { + if idx < 0 || idx >= len(h.shards) { + return nil, fmt.Errorf("shard index %d out of range: handler has %d shard(s)", idx, len(h.shards)) + } + return h.shards[idx], nil +} + type activeRequest struct { req jsonrpc.Request[json.RawMessage] labels requestLabels + shard *gwhandlers.ShardEndpoint responses map[string]*jsonrpc.Response[json.RawMessage] mu sync.Mutex completed atomic.Bool @@ -310,15 +343,6 @@ func NewHandler(methodConfig json.RawMessage, dons *gwhandlers.ShardedDONs, lggr }, nil } -// totalNodes is the number of relay nodes across every shard, used only for logging. -func (h *handler) totalNodes() int { - n := 0 - for _, shard := range h.shards { - n += len(shard.Members) - } - return n -} - func (h *handler) Start(_ context.Context) error { return h.StartOnce("ConfidentialRelayHandler", func() error { h.lggr.Info("starting confidential relay handler") @@ -370,16 +394,16 @@ func (h *handler) removeExpiredRequests(ctx context.Context) { for _, er := range expiredRequests { l := h.requestLogger(er.req, er.labels) collected := len(er.copiedResponses()) + nodes := len(er.shard.Members) l.Debugw("request expired, evaluating collected relay responses", "collected", collected, - "nodes", h.totalNodes(), - "unanswered", h.totalNodes()-collected, + "nodes", nodes, + "unanswered", nodes-collected, ) // Expiry makes further responses unavailable to this request. The common - // readiness path forwards a viable partial bundle from whichever shard - // qualifies, or returns a timeout. - if err := h.evaluateShards(ctx, l, er, true); err != nil { - l.Errorw("error evaluating relay response bundles on expiry", "error", err) + // readiness path forwards a viable partial bundle, or returns a timeout. + if err := h.evaluateShard(ctx, l, er, true); err != nil { + l.Errorw("error evaluating relay response bundle on expiry", "error", err) } } } @@ -414,11 +438,20 @@ func (h *handler) HandleJSONRPCUserMessage(ctx context.Context, req jsonrpc.Requ return errors.New("request ID is too long: " + strconv.Itoa(len(req.ID)) + ". max is 200 characters") } + shardIndex, err := extractShardIndex(req) + if err != nil { + return err + } + shard, err := h.shardAt(shardIndex) + if err != nil { + return err + } + labels := h.extractRequestLabels(req) l := h.requestLogger(req, labels) - l.Debugw("handling confidential relay request", "shards", len(h.shards), "nodes", h.totalNodes()) + l.Debugw("handling confidential relay request", "shard", shard.DonID, "nodes", len(shard.Members)) - ar, err := h.newActiveRequest(req, labels, callback) + ar, err := h.newActiveRequest(req, labels, shard, callback) if err != nil { return err } @@ -426,7 +459,7 @@ func (h *handler) HandleJSONRPCUserMessage(ctx context.Context, req jsonrpc.Requ return h.fanOutToNodes(ctx, l, ar) } -func (h *handler) newActiveRequest(req jsonrpc.Request[json.RawMessage], labels requestLabels, callback gwhandlers.Callback) (*activeRequest, error) { +func (h *handler) newActiveRequest(req jsonrpc.Request[json.RawMessage], labels requestLabels, shard *gwhandlers.ShardEndpoint, callback gwhandlers.Callback) (*activeRequest, error) { h.mu.Lock() defer h.mu.Unlock() if h.activeRequests[req.ID] != nil { @@ -437,6 +470,7 @@ func (h *handler) newActiveRequest(req jsonrpc.Request[json.RawMessage], labels Callback: callback, req: req, labels: labels, + shard: shard, createdAt: h.clock.Now(), responses: map[string]*jsonrpc.Response[json.RawMessage]{}, } @@ -482,7 +516,7 @@ func (h *handler) HandleNodeMessage(ctx context.Context, resp *jsonrpc.Response[ return nil } - return h.evaluateShards(ctx, l, ar, false) + return h.evaluateShard(ctx, l, ar, false) } // filterResponses restricts a request's collected responses to one shard's members. @@ -498,135 +532,121 @@ func filterResponses(all map[string]jsonrpc.Response[json.RawMessage], shard *gw return out } -// evaluateShards applies the completion rules to live and expired requests, once per -// shard: each shard's quorum is evaluated against its own F and members, independently -// of every other shard, and the first shard whose bundle is ready wins the request. -// Only when every shard can no longer reach its own quorum does the request fail. -// Expiry makes further responses unavailable, so remaining is forced to zero for every -// shard rather than counted. -func (h *handler) evaluateShards(ctx context.Context, l logger.Logger, ar *activeRequest, expired bool) error { - all := ar.copiedResponses() +// evaluateShard applies the same completion rules to live and expired requests, using +// the single shard this request was routed to. Remaining counts responses that can +// still arrive; expiry passes zero because waiting can no longer improve the bundle. +func (h *handler) evaluateShard(ctx context.Context, l logger.Logger, ar *activeRequest, expired bool) error { + shard := ar.shard + summary, err := h.bundler.Bundle(ar.req, filterResponses(ar.copiedResponses(), shard), l) + if err != nil { + l.Errorw("failed to build relay response bundle", "error", err) + return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.FatalError, err)) + } - var ( - allDead = true - firstUserErr *jsonrpc.WireError - totalSigned int - totalCollected int - ) + earlyNeed := 2*shard.F + 1 + minQuorum := shard.F + 1 + nodes := len(shard.Members) + unanswered := nodes - summary.Total() + remaining := unanswered + if expired { + remaining = 0 + } + maxPossibleSigned := summary.Signed() + remaining - for _, shard := range h.shards { - summary, err := h.bundler.Bundle(ar.req, filterResponses(all, shard), l) - if err != nil { - l.Errorw("failed to build relay response bundle", "shard", shard.DonID, "error", err) - return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.FatalError, err)) + if summary.Signed() >= earlyNeed { + return h.forwardBundle(ctx, l, ar, shard, summary) + } + if remaining == 0 && summary.Signed() >= minQuorum { + l.Infow("no more responses expected; forwarding partial signed bundle", + "signed", summary.Signed(), + "minQuorum", minQuorum, + "earlyNeed", earlyNeed, + "collected", summary.Total(), + "nodes", nodes, + "unanswered", unanswered, + "expired", expired, + "errors", summary.Error(), + "undecodable", summary.Undecodable(), + "nodeErrors", summary.NodeErrorsFormatted(), + ) + return h.forwardBundle(ctx, l, ar, shard, summary) + } + if maxPossibleSigned < minQuorum { + // A user-level node error means the request is invalid for every node, + // so report it rather than the generic quorum failure. + if userErr := summary.UserError(); userErr != nil { + l.Warnw("relay quorum unreachable due to a user error; propagating node error", + "signed", summary.Signed(), + "minQuorum", minQuorum, + "collected", summary.Total(), + "nodes", nodes, + "remaining", remaining, + "expired", expired, + "errors", summary.Error(), + "undecodable", summary.Undecodable(), + "userErrorCode", userErr.Code, + "nodeErrors", summary.NodeErrorsFormatted(), + ) + return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.InvalidParamsError, + errors.New(sanitizeNodeErrorMessage(userErr.Message)))) } - - minQuorum := shard.F + 1 - earlyNeed := 2*shard.F + 1 - nodes := len(shard.Members) - remaining := nodes - summary.Total() if expired { - remaining = 0 - } - maxPossibleSigned := summary.Signed() + remaining - - totalSigned += summary.Signed() - totalCollected += summary.Total() - - if summary.Signed() >= earlyNeed { - return h.forwardBundle(ctx, l, ar, shard, summary) - } - if remaining == 0 && summary.Signed() >= minQuorum { - l.Infow("no more responses expected in shard; forwarding partial signed bundle", - "shard", shard.DonID, + l.Warnw("request expired before relay quorum was reached", "signed", summary.Signed(), "minQuorum", minQuorum, "earlyNeed", earlyNeed, "collected", summary.Total(), "nodes", nodes, - "unanswered", nodes-summary.Total(), - "expired", expired, + "unanswered", unanswered, "errors", summary.Error(), "undecodable", summary.Undecodable(), "nodeErrors", summary.NodeErrorsFormatted(), ) - return h.forwardBundle(ctx, l, ar, shard, summary) - } - if maxPossibleSigned < minQuorum { - // A user-level node error means the request is invalid for every node, - // so report it rather than the generic quorum failure. - if userErr := summary.UserError(); userErr != nil && firstUserErr == nil { - firstUserErr = userErr - } - continue + return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.RequestTimeoutError, + fmt.Errorf("request expired: %d signed relay responses, need at least %d to reach quorum (collected=%d nodes=%d unanswered=%d errors=%d undecodable=%d)", + summary.Signed(), minQuorum, summary.Total(), nodes, unanswered, summary.Error(), summary.Undecodable()))) } - - allDead = false - if !expired && summary.Signed() >= minQuorum { - h.startQuorumGrace(l, ar, summary, shard) - } - } - - if allDead { - return h.failUnreachableQuorum(ctx, l, ar, firstUserErr, expired, totalSigned, totalCollected) - } - - l.Debugw("waiting for more signed relay responses before forwarding bundle", - "signed", totalSigned, - "collected", totalCollected, - "nodes", h.totalNodes(), - "shards", len(h.shards), - ) - return nil -} - -// failUnreachableQuorum is reached only once every shard's quorum has become -// unreachable: a single-shard "quorum unreachable" verdict no longer fails the -// request outright, since another shard may still complete it. -func (h *handler) failUnreachableQuorum(ctx context.Context, l logger.Logger, ar *activeRequest, userErr *jsonrpc.WireError, expired bool, signed, collected int) error { - nodes := h.totalNodes() - if userErr != nil { - l.Warnw("relay quorum unreachable in every shard due to a user error; propagating node error", - "signed", signed, - "collected", collected, + l.Warnw("relay quorum unreachable; returning error without waiting", + "signed", summary.Signed(), + "minQuorum", minQuorum, + "earlyNeed", earlyNeed, + "collected", summary.Total(), "nodes", nodes, - "shards", len(h.shards), - "expired", expired, - "userErrorCode", userErr.Code, + "remaining", remaining, + "maxPossibleSigned", maxPossibleSigned, + "errors", summary.Error(), + "undecodable", summary.Undecodable(), + "nodeErrors", summary.NodeErrorsFormatted(), ) - return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.InvalidParamsError, - errors.New(sanitizeNodeErrorMessage(userErr.Message)))) + return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.FatalError, + fmt.Errorf("relay quorum unreachable: %d signed responses, at most %d possible, need %d (collected=%d nodes=%d remaining=%d errors=%d undecodable=%d)", + summary.Signed(), maxPossibleSigned, minQuorum, summary.Total(), nodes, remaining, summary.Error(), summary.Undecodable()))) } - if expired { - l.Warnw("request expired before relay quorum was reached in any shard", - "signed", signed, - "collected", collected, - "nodes", nodes, - "shards", len(h.shards), - ) - return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.RequestTimeoutError, - fmt.Errorf("request expired: %d signed relay responses across %d shard(s), none reached quorum (collected=%d nodes=%d)", - signed, len(h.shards), collected, nodes))) + if !expired && summary.Signed() >= minQuorum { + h.startQuorumGrace(l, ar, summary, shard) } - l.Warnw("relay quorum unreachable in every shard; returning error without waiting", - "signed", signed, - "collected", collected, + l.Debugw("waiting for more signed relay responses before forwarding bundle", + "signed", summary.Signed(), + "earlyNeed", earlyNeed, + "minQuorum", minQuorum, + "collected", summary.Total(), "nodes", nodes, - "shards", len(h.shards), + "remaining", remaining, + "maxPossibleSigned", maxPossibleSigned, + "errors", summary.Error(), + "undecodable", summary.Undecodable(), + "nodeErrors", summary.NodeErrorsFormatted(), ) - return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.FatalError, - fmt.Errorf("relay quorum unreachable: %d signed responses across %d shard(s), none can reach quorum (collected=%d nodes=%d)", - signed, len(h.shards), collected, nodes))) + return nil } // startQuorumGrace arms the grace window the first time a request holds minQuorum -// signed responses in some shard. It bounds how long a request that will never reach -// earlyNeed keeps waiting for the rest of the DON: without it such a request is held -// until requestTimeout, long after the caller's own deadline has elapsed, so a bundle +// signed responses. It bounds how long a request that will never reach earlyNeed +// keeps waiting for the rest of the DON: without it such a request is held until +// requestTimeout, long after the caller's own deadline has elapsed, so a bundle // that was viable within milliseconds is forwarded to nobody. The deadline is swept // by the cleanup goroutine rather than a timer, so no completion path runs on an -// untracked goroutine. It is armed once per request (not per shard): the first shard -// to reach minQuorum starts the clock for whichever shard ends up winning. +// untracked goroutine. func (h *handler) startQuorumGrace(l logger.Logger, ar *activeRequest, summary *BundleSummary, shard *gwhandlers.ShardEndpoint) { if h.quorumGrace <= 0 { return @@ -666,44 +686,43 @@ func (h *handler) forwardGracedRequests(ctx context.Context) { func (h *handler) forwardAfterGrace(ctx context.Context, ar *activeRequest) { l := h.requestLogger(ar.req, ar.labels) - all := ar.copiedResponses() - - for _, shard := range h.shards { - summary, err := h.bundler.Bundle(ar.req, filterResponses(all, shard), l) - if err != nil { - l.Errorw("failed to build relay response bundle after quorum grace", "shard", shard.DonID, "error", err) - if sendErr := h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.FatalError, err)); sendErr != nil { - l.Errorw("error returning bundle failure after quorum grace", "error", sendErr) - } - return - } - - minQuorum := shard.F + 1 - if summary.Signed() < minQuorum { - continue + shard := ar.shard + summary, err := h.bundler.Bundle(ar.req, filterResponses(ar.copiedResponses(), shard), l) + if err != nil { + l.Errorw("failed to build relay response bundle after quorum grace", "error", err) + if sendErr := h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.FatalError, err)); sendErr != nil { + l.Errorw("error returning bundle failure after quorum grace", "error", sendErr) } + return + } - nodes := len(shard.Members) - l.Infow("quorum grace elapsed; forwarding partial signed bundle", - "shard", shard.DonID, + minQuorum := shard.F + 1 + nodes := len(shard.Members) + if summary.Signed() < minQuorum { + l.Warnw("quorum grace elapsed below quorum; leaving request to expiry", "signed", summary.Signed(), "minQuorum", minQuorum, - "earlyNeed", 2*shard.F+1, "collected", summary.Total(), "nodes", nodes, - "unanswered", nodes-summary.Total(), - "grace", h.quorumGrace, - "errors", summary.Error(), - "undecodable", summary.Undecodable(), - "nodeErrors", summary.NodeErrorsFormatted(), ) - if err := h.forwardBundle(ctx, l, ar, shard, summary); err != nil { - l.Errorw("error forwarding bundle after quorum grace", "error", err) - } return } - l.Warnw("quorum grace elapsed without any shard reaching quorum; leaving request to expiry") + l.Infow("quorum grace elapsed; forwarding partial signed bundle", + "signed", summary.Signed(), + "minQuorum", minQuorum, + "earlyNeed", 2*shard.F+1, + "collected", summary.Total(), + "nodes", nodes, + "unanswered", nodes-summary.Total(), + "grace", h.quorumGrace, + "errors", summary.Error(), + "undecodable", summary.Undecodable(), + "nodeErrors", summary.NodeErrorsFormatted(), + ) + if err := h.forwardBundle(ctx, l, ar, shard, summary); err != nil { + l.Errorw("error forwarding bundle after quorum grace", "error", err) + } } // forwardBundle sends a previously-built shard bundle to the enclave. The gateway makes @@ -732,13 +751,14 @@ func (h *handler) forwardBundle(ctx context.Context, l logger.Logger, ar *active }) } -// fanOutToNodes sends the request to every member of every shard, each via that -// shard's own connection manager. A shard whose own quorum becomes unreachable here -// does not fail the request by itself: another shard may still complete it. The -// request only fails if every shard is left unreachable. +// fanOutToNodes sends the request to every member of the shard this request was +// routed to, via that shard's own connection manager. func (h *handler) fanOutToNodes(ctx context.Context, l logger.Logger, ar *activeRequest) error { - var group errgroup.Group - shardNodeErrors := make([]atomic.Uint32, len(h.shards)) + var ( + group errgroup.Group + nodeErrors atomic.Uint32 + ) + shard := ar.shard // Each send is bounded independently. A node whose websocket accepts no writes blocks // until its context is cancelled, and because the caller only reads the response callback @@ -747,47 +767,34 @@ func (h *handler) fanOutToNodes(ctx context.Context, l logger.Logger, ar *active sendCtx, cancel := context.WithTimeout(ctx, h.nodeSendTimeout) defer cancel() - for i, shard := range h.shards { - for _, node := range shard.Members { - group.Go(func() error { - err := shard.ConnMgr.SendToNode(sendCtx, node.Address, &ar.req) - if err != nil { - shardNodeErrors[i].Add(1) - l.Errorw("error sending request to node", "node", node.Address, "shard", shard.DonID, "error", err) - } - return nil - }) - } + for _, node := range shard.Members { + group.Go(func() error { + err := shard.ConnMgr.SendToNode(sendCtx, node.Address, &ar.req) + if err != nil { + nodeErrors.Add(1) + l.Errorw("error sending request to node", "node", node.Address, "error", err) + } + return nil + }) } _ = group.Wait() - var anyShardReachable bool - var totalNodeErrors uint32 - for i, shard := range h.shards { - numNodeErrors := shardNodeErrors[i].Load() - totalNodeErrors += numNodeErrors - reachable := len(shard.Members) - int(numNodeErrors) - if reachable >= shard.F+1 { - anyShardReachable = true - } else if numNodeErrors > 0 { - l.Errorw("too few relay nodes reachable in shard to reach quorum", - "shard", shard.DonID, - "nodeErrors", numNodeErrors, - "reachable", reachable, - "minQuorum", shard.F+1, - "nodes", len(shard.Members), - ) - } - } - if !anyShardReachable { - // The per-shard failures are logged above; this is the aggregate decision - // that ends the request, so it gets its own line. - l.Errorw("too few relay nodes reachable in any shard to reach quorum; failing request", "nodeErrors", totalNodeErrors) + numNodeErrors := nodeErrors.Load() + remainingPossibleResponses := len(shard.Members) - int(numNodeErrors) + if remainingPossibleResponses < shard.F+1 && numNodeErrors > 0 { + // The individual send failures are logged above; this is the aggregate + // decision that ends the request, so it gets its own line. + l.Errorw("too few relay nodes reachable to reach quorum; failing request", + "nodeErrors", numNodeErrors, + "reachable", remainingPossibleResponses, + "minQuorum", shard.F+1, + "nodes", len(shard.Members), + ) return h.sendResponseAndClearRequest(ctx, ar, h.constructErrorResponse(ar.req, api.FatalError, errors.New("failed to forward user request to nodes"))) } - l.Debugw("successfully forwarded request to relay nodes", "nodeErrors", totalNodeErrors) + l.Debugw("successfully forwarded request to relay nodes", "nodeErrors", numNodeErrors) return nil } @@ -813,26 +820,27 @@ func (h *handler) sendResponseAndClearRequest(ctx context.Context, ar *activeReq return sendErr } - h.recordMetrics(ctx, payload.ErrorCode) + h.recordMetrics(ctx, ar, payload.ErrorCode) h.lggr.Debugw("response sent to user", "requestID", ar.req.ID, "executionID", ar.labels.ExecutionID, "errorCode", payload.ErrorCode) return nil } -func (h *handler) recordMetrics(ctx context.Context, errorCode api.ErrorCode) { +func (h *handler) recordMetrics(ctx context.Context, ar *activeRequest, errorCode api.ErrorCode) { + donID := ar.shard.DonID //nolint:exhaustive // do not record other errors switch errorCode { case api.HandlerError: h.metrics.requestInternalError.Add(ctx, 1, metric.WithAttributes( - attribute.String("don_id", h.donID), + attribute.String("don_id", donID), attribute.String("error", errorCode.String()), )) case api.UnsupportedDONIdError: h.metrics.requestUserError.Add(ctx, 1, metric.WithAttributes( - attribute.String("don_id", h.donID), + attribute.String("don_id", donID), )) case api.NoError: h.metrics.requestSuccess.Add(ctx, 1, metric.WithAttributes( - attribute.String("don_id", h.donID), + attribute.String("don_id", donID), )) } } diff --git a/core/services/gateway/handlers/confidentialrelay/handler_test.go b/core/services/gateway/handlers/confidentialrelay/handler_test.go index 3b82bfb9ca8..a229b2a8b95 100644 --- a/core/services/gateway/handlers/confidentialrelay/handler_test.go +++ b/core/services/gateway/handlers/confidentialrelay/handler_test.go @@ -883,7 +883,7 @@ func TestConfidentialRelayHandler_DuplicateRequestID(t *testing.T) { h, cb, don, _ := setupHandler(t, 4) don.On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(nil) - params := json.RawMessage(`{"workflow_id":"wf1"}`) + params := json.RawMessage(`{"workflow_id":"wf1","shard_index":0}`) req := jsonrpc.Request[json.RawMessage]{ ID: "req-dup", Method: MethodCapabilityExec, @@ -925,7 +925,7 @@ func TestConfidentialRelayHandler_RateLimitedNode(t *testing.T) { don.On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(nil) cb := common.NewCallback() - params := json.RawMessage(`{"workflow_id":"wf1"}`) + params := json.RawMessage(`{"workflow_id":"wf1","shard_index":0}`) req := jsonrpc.Request[json.RawMessage]{ ID: "req-ratelimit", Method: MethodCapabilityExec, @@ -1010,7 +1010,7 @@ func TestConfidentialRelayHandler_AllNodesFanOutFail(t *testing.T) { h, cb, don, _ := setupHandler(t, 4) don.On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(errors.New("connection refused")) - params := json.RawMessage(`{"workflow_id":"wf1"}`) + params := json.RawMessage(`{"workflow_id":"wf1","shard_index":0}`) req := jsonrpc.Request[json.RawMessage]{ ID: "req-allfail", Method: MethodCapabilityExec, @@ -1051,7 +1051,7 @@ func TestConfidentialRelayHandler_FanOutWaitsWhileQuorumStillPossible(t *testing }, ) - params := json.RawMessage(`{"workflow_id":"wf1"}`) + params := json.RawMessage(`{"workflow_id":"wf1","shard_index":0}`) req := jsonrpc.Request[json.RawMessage]{ ID: "req-still-possible", Method: MethodCapabilityExec, @@ -1083,7 +1083,7 @@ func TestConfidentialRelayHandler_FanOutFailsWhenQuorumBecomesImpossible(t *test }, ) - params := json.RawMessage(`{"workflow_id":"wf1"}`) + params := json.RawMessage(`{"workflow_id":"wf1","shard_index":0}`) req := jsonrpc.Request[json.RawMessage]{ ID: "req-quorum-impossible", Method: MethodCapabilityExec, @@ -1135,7 +1135,7 @@ func TestConfidentialRelayHandler_FanOutToNodes_IsConcurrent(t *testing.T) { require.NoError(t, err) cb := common.NewCallback() - params := json.RawMessage(`{"workflow_id":"wf1"}`) + params := json.RawMessage(`{"workflow_id":"wf1","shard_index":0}`) req := jsonrpc.Request[json.RawMessage]{ ID: "req-concurrent-fanout", Method: MethodCapabilityExec, @@ -1220,7 +1220,7 @@ func TestConfidentialRelayHandler_BlockedNodeDoesNotStallFanOut(t *testing.T) { // Shortened so the test exercises the bound without waiting the production default. h.nodeSendTimeout = 50 * time.Millisecond - params := json.RawMessage(`{"workflow_id":"wf1"}`) + params := json.RawMessage(`{"workflow_id":"wf1","shard_index":0}`) req := jsonrpc.Request[json.RawMessage]{ ID: "req-blocked-node", Method: MethodCapabilityExec, @@ -1293,27 +1293,27 @@ func setupMultiShardHandler(t *testing.T, f int, shardSizes ...int) (*handler, * return h, cb, mockConns, clock } -func nodeErrorResponse(id string) *jsonrpc.Response[json.RawMessage] { - return &jsonrpc.Response[json.RawMessage]{ - Version: jsonrpc.JsonRpcVersion, - ID: id, - Method: MethodCapabilityExec, - Error: &jsonrpc.WireError{Code: -32603, Message: "internal error"}, - } +// capParamsForShard returns capability-exec params targeting a specific shard index. +func capParamsForShard(t *testing.T, workflowID string, shardIndex int) json.RawMessage { + t.Helper() + params := validCapParams(workflowID) + params.ShardIndex = new(shardIndex) + raw, err := json.Marshal(params) + require.NoError(t, err) + return raw } -// First shard to reach quorum wins: shard 0 reaches 2F+1 while shard 1 has no -// responses at all, and the forwarded bundle contains only shard 0's responses. -func TestConfidentialRelayHandler_Sharded_FirstShardToReachQuorumWins(t *testing.T) { +// A request naming shard 1 fans out only to shard 1's connection manager and ignores +// shard 0 entirely: shard 0's mock DON has no expectations set, so mockery fails the +// test if fan-out ever reaches it. +func TestConfidentialRelayHandler_Sharded_RoutesToSelectedShardOnly(t *testing.T) { t.Parallel() h, cb, conns, _ := setupMultiShardHandler(t, 1, 4, 4) // F=1: minQuorum=2, earlyNeed=3 per shard - for _, conn := range conns { - conn.On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(nil) - } + conns[1].On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(nil) - params := validCapParamsJSON("wf1") + params := capParamsForShard(t, "wf1", 1) req := jsonrpc.Request[json.RawMessage]{ - ID: "req-sharded-win", + ID: "req-sharded-routed", Method: MethodCapabilityExec, Params: ¶ms, } @@ -1336,105 +1336,51 @@ func TestConfidentialRelayHandler_Sharded_FirstShardToReachQuorumWins(t *testing if !assert.NoError(t, json.Unmarshal(*jsonResp.Result, &bundle)) { return } - assert.Len(t, bundle.Responses, 3, "only shard 0's signed responses are bundled") + assert.Len(t, bundle.Responses, 3) }) - // Shard 0 is nodes 0x0000-0x0003; reaching 3 signed responses hits 2F+1 there. - for i := range 3 { + // A response from a shard 0 node address must not count toward this request's + // quorum, even though it shares a rate limiter map with shard 1's nodes. + require.NoError(t, h.HandleNodeMessage(t.Context(), + capExecSignedRespPtr(t, req.ID, result, []byte("signer-foreign")), "0x0000")) + require.NotNil(t, h.getActiveRequest(req.ID), "a foreign-shard response must not complete the request") + + // Shard 1 is nodes 0x0004-0x0007; reaching 3 signed responses hits 2F+1 there. + for i := 4; i < 7; i++ { require.NoError(t, h.HandleNodeMessage(t.Context(), capExecSignedRespPtr(t, req.ID, result, fmt.Appendf(nil, "signer-%d", i)), fmt.Sprintf("0x%04d", i))) } wg.Wait() - require.Nil(t, h.getActiveRequest(req.ID), "request completes once any shard reaches quorum") - - // A late response from shard 1 (nodes 0x0004-0x0007) must be ignored: the request - // is already answered, and shard 1's quorum is irrelevant once shard 0 has won. - require.NoError(t, h.HandleNodeMessage(t.Context(), - capExecSignedRespPtr(t, req.ID, result, []byte("signer-late")), "0x0004")) + require.Nil(t, h.getActiveRequest(req.ID)) } -// A shard whose own quorum becomes unreachable does not fail the request: another -// shard can still complete it independently. -func TestConfidentialRelayHandler_Sharded_DeadShardDoesNotFailRequest(t *testing.T) { +func TestConfidentialRelayHandler_ShardIndexMissing(t *testing.T) { t.Parallel() - h, cb, conns, _ := setupMultiShardHandler(t, 1, 4, 4) - for _, conn := range conns { - conn.On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(nil) - } + h, cb, _, _ := setupHandler(t, 4) - params := validCapParamsJSON("wf1") + params := json.RawMessage(`{"workflow_id":"wf1"}`) req := jsonrpc.Request[json.RawMessage]{ - ID: "req-sharded-dead", + ID: "req-no-shard", Method: MethodCapabilityExec, Params: ¶ms, } - result := relaytypes.CapabilityResponseResult{Payload: "result"} - - require.NoError(t, h.HandleJSONRPCUserMessage(t.Context(), req, cb)) - - // Shard 0 (nodes 0x0000-0x0003) answers entirely with errors: once every member - // has replied, signed stays 0 and this shard can never reach minQuorum=2. - for i := range 4 { - require.NoError(t, h.HandleNodeMessage(t.Context(), nodeErrorResponse(req.ID), fmt.Sprintf("0x%04d", i))) - } - require.NotNil(t, h.getActiveRequest(req.ID), "shard 1 can still reach quorum") - var wg sync.WaitGroup - wg.Go(func() { - resp, err := cb.Wait(t.Context()) - if !assert.NoError(t, err) { //nolint:testifylint // require illegal inside wg.Go goroutine - return - } - assert.Equal(t, api.NoError, resp.ErrorCode) - }) - - // Shard 1 (nodes 0x0004-0x0007) reaches 2F+1=3 signed responses. - for i := 4; i < 7; i++ { - require.NoError(t, h.HandleNodeMessage(t.Context(), - capExecSignedRespPtr(t, req.ID, result, fmt.Appendf(nil, "signer-%d", i)), - fmt.Sprintf("0x%04d", i))) - } - wg.Wait() - require.Nil(t, h.getActiveRequest(req.ID)) + err := h.HandleJSONRPCUserMessage(t.Context(), req, cb) + require.EqualError(t, err, "shard_index is required") } -// Only once every shard's quorum is unreachable does the request fail. -func TestConfidentialRelayHandler_Sharded_AllShardsDeadFailsRequest(t *testing.T) { +func TestConfidentialRelayHandler_ShardIndexOutOfRange(t *testing.T) { t.Parallel() - h, cb, conns, _ := setupMultiShardHandler(t, 1, 4, 4) - for _, conn := range conns { - conn.On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(nil) - } + h, cb, _, _ := setupHandler(t, 4) - params := validCapParamsJSON("wf1") + params := capParamsForShard(t, "wf1", 1) req := jsonrpc.Request[json.RawMessage]{ - ID: "req-sharded-all-dead", + ID: "req-shard-oor", Method: MethodCapabilityExec, Params: ¶ms, } - var wg sync.WaitGroup - wg.Go(func() { - resp, err := cb.Wait(t.Context()) - if !assert.NoError(t, err) { //nolint:testifylint // require illegal inside wg.Go goroutine - return - } - assert.Equal(t, api.FatalError, resp.ErrorCode) - var jsonResp jsonrpc.Response[json.RawMessage] - if !assert.NoError(t, json.Unmarshal(resp.RawResponse, &jsonResp)) { //nolint:testifylint // require illegal inside wg.Go goroutine - return - } - if assert.NotNil(t, jsonResp.Error) { - assert.Contains(t, jsonResp.Error.Message, "relay quorum unreachable") - } - }) - - require.NoError(t, h.HandleJSONRPCUserMessage(t.Context(), req, cb)) - - for i := range 8 { - require.NoError(t, h.HandleNodeMessage(t.Context(), nodeErrorResponse(req.ID), fmt.Sprintf("0x%04d", i))) - } - wg.Wait() - require.Nil(t, h.getActiveRequest(req.ID)) + err := h.HandleJSONRPCUserMessage(t.Context(), req, cb) + require.EqualError(t, err, "shard index 1 out of range: handler has 1 shard(s)") } diff --git a/go.mod b/go.mod index 24afa53726f..278cc230fab 100644 --- a/go.mod +++ b/go.mod @@ -82,7 +82,7 @@ require ( github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260908155618-50f521d70e62 github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260908155618-50f521d70e62 github.com/smartcontractkit/chainlink-ccv v0.14.0 - github.com/smartcontractkit/chainlink-common v0.11.2-0.20261007174852-7fabc093ff84 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20261014212707-0e9847f3964e github.com/smartcontractkit/chainlink-common/keystore v1.3.1-0.20260903141829-ef07b52a737d github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549 github.com/smartcontractkit/chainlink-data-streams v1.1.2-0.20261009064810-68424c714170 diff --git a/go.sum b/go.sum index 6f71010fdfa..17b360daf27 100644 --- a/go.sum +++ b/go.sum @@ -1106,6 +1106,8 @@ github.com/smartcontractkit/chainlink-ccv v0.14.0 h1:184WuWwWAOo8b5uTuxv8v+zI7Uc github.com/smartcontractkit/chainlink-ccv v0.14.0/go.mod h1:k/RxmdtzRqqiNcl5X2Af7Q7R53ZAaMwgyd9ctMKtK6Q= github.com/smartcontractkit/chainlink-common v0.11.2-0.20261007174852-7fabc093ff84 h1:IvG1s9HGBjHtDayPBCKoeXVLsneIXEZhm5vF1X5d9bk= github.com/smartcontractkit/chainlink-common v0.11.2-0.20261007174852-7fabc093ff84/go.mod h1:37Mu0pbAnGmHPgOZ8bQvgP4hzXw/Wu7smKX4cN46UOs= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20261014212707-0e9847f3964e h1:ojGMaWZS2lW8jbnqhk+aTrnLo1zNmcHN1dfa18t9Yog= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20261014212707-0e9847f3964e/go.mod h1:37Mu0pbAnGmHPgOZ8bQvgP4hzXw/Wu7smKX4cN46UOs= github.com/smartcontractkit/chainlink-common/keystore v1.3.1-0.20260903141829-ef07b52a737d h1:M9xuiIvdiHtim4nlxCVajyC/Mk63tAg9ESWb/biAhNA= github.com/smartcontractkit/chainlink-common/keystore v1.3.1-0.20260903141829-ef07b52a737d/go.mod h1:rI2njmTlyTvh+xhmORmK6qX3+V1ycDAqGJLu8LS8WpU= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549 h1:J4vYRJxkBO8Yf//17wScKjYWmS45VZaFWx0zt+4ETEU= From d967855797b49f360655affc07c576bbe22f9a57 Mon Sep 17 00:00:00 2001 From: Bolek Kulbabinski <1416262+bolekk@users.noreply.github.com> Date: Wed, 14 Oct 2026 22:33:52 -0400 Subject: [PATCH 3/5] cre-6798: default shard_index to 0 when omitted Match chainlink-common#2470's simplified design (ShardIndex went from *int to plain int): a request without shard_index now routes to shard 0 instead of being rejected. Co-Authored-By: Claude Sonnet 5 --- .../confidentialrelay/bundler_test.go | 2 -- .../handlers/confidentialrelay/handler.go | 19 +++++------- .../confidentialrelay/handler_test.go | 31 ++++++++++++++++--- go.mod | 2 +- go.sum | 2 ++ 5 files changed, 36 insertions(+), 20 deletions(-) diff --git a/core/services/gateway/handlers/confidentialrelay/bundler_test.go b/core/services/gateway/handlers/confidentialrelay/bundler_test.go index 23370c8043c..139bbfa7350 100644 --- a/core/services/gateway/handlers/confidentialrelay/bundler_test.go +++ b/core/services/gateway/handlers/confidentialrelay/bundler_test.go @@ -30,7 +30,6 @@ func validCapParams(workflowID string) relaytypes.CapabilityRequestParams { ReferenceID: "ref-1", CapabilityID: "cap-1", Payload: "in", - ShardIndex: new(0), } } @@ -41,7 +40,6 @@ func validSecretsParams(workflowID string) relaytypes.SecretsRequestParams { ExecutionID: testExecutionID, Secrets: []relaytypes.SecretIdentifier{{Key: "k1", Namespace: "ns"}}, EnclavePublicKey: testEnclavePK, - ShardIndex: new(0), } } diff --git a/core/services/gateway/handlers/confidentialrelay/handler.go b/core/services/gateway/handlers/confidentialrelay/handler.go index e8c986a9ae3..18073772cf2 100644 --- a/core/services/gateway/handlers/confidentialrelay/handler.go +++ b/core/services/gateway/handlers/confidentialrelay/handler.go @@ -106,28 +106,23 @@ func (h *handler) extractRequestLabels(req jsonrpc.Request[json.RawMessage]) req return labels } -// requestShardSelector is the shard-routing field both relay methods' params carry. -// Unlike requestLabels, decoding this is not best-effort: a request that omits it, or -// names a shard the handler does not have, is rejected rather than silently routed to -// shard 0, since 0 is itself a valid shard and must not be indistinguishable from "not -// set." +// requestShardSelector is the shard-routing field both relay methods' params carry. A +// request that omits it defaults to shard 0. type requestShardSelector struct { - ShardIndex *int `json:"shard_index"` + ShardIndex int `json:"shard_index"` } -// extractShardIndex decodes the caller-selected shard index out of a request's params. +// extractShardIndex decodes the caller-selected shard index out of a request's params, +// defaulting to 0 if params is absent or does not carry the field. func extractShardIndex(req jsonrpc.Request[json.RawMessage]) (int, error) { if req.Params == nil { - return 0, errors.New("shard_index is required") + return 0, nil } var sel requestShardSelector if err := json.Unmarshal(*req.Params, &sel); err != nil { return 0, fmt.Errorf("could not decode shard_index: %w", err) } - if sel.ShardIndex == nil { - return 0, errors.New("shard_index is required") - } - return *sel.ShardIndex, nil + return sel.ShardIndex, nil } // shardAt resolves a caller-selected shard index to the shard it names. diff --git a/core/services/gateway/handlers/confidentialrelay/handler_test.go b/core/services/gateway/handlers/confidentialrelay/handler_test.go index a229b2a8b95..ba094477bfe 100644 --- a/core/services/gateway/handlers/confidentialrelay/handler_test.go +++ b/core/services/gateway/handlers/confidentialrelay/handler_test.go @@ -1297,7 +1297,7 @@ func setupMultiShardHandler(t *testing.T, f int, shardSizes ...int) (*handler, * func capParamsForShard(t *testing.T, workflowID string, shardIndex int) json.RawMessage { t.Helper() params := validCapParams(workflowID) - params.ShardIndex = new(shardIndex) + params.ShardIndex = shardIndex raw, err := json.Marshal(params) require.NoError(t, err) return raw @@ -1355,9 +1355,12 @@ func TestConfidentialRelayHandler_Sharded_RoutesToSelectedShardOnly(t *testing.T require.Nil(t, h.getActiveRequest(req.ID)) } -func TestConfidentialRelayHandler_ShardIndexMissing(t *testing.T) { +// A request that omits shard_index defaults to shard 0: fan-out reaches only shard 0's +// connection manager, never shard 1's. +func TestConfidentialRelayHandler_ShardIndexMissingDefaultsToShardZero(t *testing.T) { t.Parallel() - h, cb, _, _ := setupHandler(t, 4) + h, cb, conns, _ := setupMultiShardHandler(t, 1, 4, 4) + conns[0].On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(nil) params := json.RawMessage(`{"workflow_id":"wf1"}`) req := jsonrpc.Request[json.RawMessage]{ @@ -1366,8 +1369,26 @@ func TestConfidentialRelayHandler_ShardIndexMissing(t *testing.T) { Params: ¶ms, } - err := h.HandleJSONRPCUserMessage(t.Context(), req, cb) - require.EqualError(t, err, "shard_index is required") + require.NoError(t, h.HandleJSONRPCUserMessage(t.Context(), req, cb)) + + result := relaytypes.CapabilityResponseResult{Payload: "result"} + var wg sync.WaitGroup + wg.Go(func() { + resp, err := cb.Wait(t.Context()) + if !assert.NoError(t, err) { //nolint:testifylint // require illegal inside wg.Go goroutine + return + } + assert.Equal(t, api.NoError, resp.ErrorCode) + }) + + // Shard 0 is nodes 0x0000-0x0003; reaching 3 signed responses hits 2F+1 there. + for i := range 3 { + require.NoError(t, h.HandleNodeMessage(t.Context(), + capExecSignedRespPtr(t, req.ID, result, fmt.Appendf(nil, "signer-%d", i)), + fmt.Sprintf("0x%04d", i))) + } + wg.Wait() + require.Nil(t, h.getActiveRequest(req.ID)) } func TestConfidentialRelayHandler_ShardIndexOutOfRange(t *testing.T) { diff --git a/go.mod b/go.mod index 278cc230fab..f0bb2d12da8 100644 --- a/go.mod +++ b/go.mod @@ -82,7 +82,7 @@ require ( github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260908155618-50f521d70e62 github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260908155618-50f521d70e62 github.com/smartcontractkit/chainlink-ccv v0.14.0 - github.com/smartcontractkit/chainlink-common v0.11.2-0.20261014212707-0e9847f3964e + github.com/smartcontractkit/chainlink-common v0.11.2-0.20261015023017-0cf17e19699d github.com/smartcontractkit/chainlink-common/keystore v1.3.1-0.20260903141829-ef07b52a737d github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549 github.com/smartcontractkit/chainlink-data-streams v1.1.2-0.20261009064810-68424c714170 diff --git a/go.sum b/go.sum index 17b360daf27..4ceec8c135b 100644 --- a/go.sum +++ b/go.sum @@ -1108,6 +1108,8 @@ github.com/smartcontractkit/chainlink-common v0.11.2-0.20261007174852-7fabc093ff github.com/smartcontractkit/chainlink-common v0.11.2-0.20261007174852-7fabc093ff84/go.mod h1:37Mu0pbAnGmHPgOZ8bQvgP4hzXw/Wu7smKX4cN46UOs= github.com/smartcontractkit/chainlink-common v0.11.2-0.20261014212707-0e9847f3964e h1:ojGMaWZS2lW8jbnqhk+aTrnLo1zNmcHN1dfa18t9Yog= github.com/smartcontractkit/chainlink-common v0.11.2-0.20261014212707-0e9847f3964e/go.mod h1:37Mu0pbAnGmHPgOZ8bQvgP4hzXw/Wu7smKX4cN46UOs= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20261015023017-0cf17e19699d h1:gwdUsOgsof08OhknvnEMM/PGfjahCOZNfhCDeffyfi4= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20261015023017-0cf17e19699d/go.mod h1:37Mu0pbAnGmHPgOZ8bQvgP4hzXw/Wu7smKX4cN46UOs= github.com/smartcontractkit/chainlink-common/keystore v1.3.1-0.20260903141829-ef07b52a737d h1:M9xuiIvdiHtim4nlxCVajyC/Mk63tAg9ESWb/biAhNA= github.com/smartcontractkit/chainlink-common/keystore v1.3.1-0.20260903141829-ef07b52a737d/go.mod h1:rI2njmTlyTvh+xhmORmK6qX3+V1ycDAqGJLu8LS8WpU= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549 h1:J4vYRJxkBO8Yf//17wScKjYWmS45VZaFWx0zt+4ETEU= From fc65e56f6909d4890606ef957b43e5d563c2e400 Mon Sep 17 00:00:00 2001 From: Bolek Kulbabinski <1416262+bolekk@users.noreply.github.com> Date: Thu, 15 Oct 2026 23:04:40 -0400 Subject: [PATCH 4/5] cre-6798: select relay shard by don_name and shard_index A gateway service can attach several DONs, so shard_index alone was ambiguous. The handler now picks the DON named by don_name (optional when there is a single DON) and the shard_index within it. Bumps chainlink-common to the branch adding DonName (#2476). Co-Authored-By: Claude Sonnet 5.5 --- .../handlers/confidentialrelay/handler.go | 75 ++++++++++++------- .../confidentialrelay/handler_test.go | 69 ++++++++++++++++- go.mod | 4 +- go.sum | 4 + 4 files changed, 124 insertions(+), 28 deletions(-) diff --git a/core/services/gateway/handlers/confidentialrelay/handler.go b/core/services/gateway/handlers/confidentialrelay/handler.go index 18073772cf2..0a9601b8e00 100644 --- a/core/services/gateway/handlers/confidentialrelay/handler.go +++ b/core/services/gateway/handlers/confidentialrelay/handler.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "strconv" "sync" "sync/atomic" @@ -106,31 +107,47 @@ func (h *handler) extractRequestLabels(req jsonrpc.Request[json.RawMessage]) req return labels } -// requestShardSelector is the shard-routing field both relay methods' params carry. A -// request that omits it defaults to shard 0. +// requestShardSelector is the routing fields both relay methods' params carry. A request +// that omits shard_index defaults to shard 0; one that omits don_name targets the only DON. type requestShardSelector struct { - ShardIndex int `json:"shard_index"` + DonName string `json:"don_name"` + ShardIndex int `json:"shard_index"` } -// extractShardIndex decodes the caller-selected shard index out of a request's params, -// defaulting to 0 if params is absent or does not carry the field. -func extractShardIndex(req jsonrpc.Request[json.RawMessage]) (int, error) { +func extractShardSelector(req jsonrpc.Request[json.RawMessage]) (requestShardSelector, error) { + var sel requestShardSelector if req.Params == nil { - return 0, nil + return sel, nil } - var sel requestShardSelector if err := json.Unmarshal(*req.Params, &sel); err != nil { - return 0, fmt.Errorf("could not decode shard_index: %w", err) + return sel, fmt.Errorf("could not decode shard selector: %w", err) } - return sel.ShardIndex, nil + return sel, nil } -// shardAt resolves a caller-selected shard index to the shard it names. -func (h *handler) shardAt(idx int) (*gwhandlers.ShardEndpoint, error) { - if idx < 0 || idx >= len(h.shards) { - return nil, fmt.Errorf("shard index %d out of range: handler has %d shard(s)", idx, len(h.shards)) +// resolveShard picks the shard named by the caller's DON name and shard index. +func (h *handler) resolveShard(sel requestShardSelector) (*gwhandlers.ShardEndpoint, error) { + donName := sel.DonName + if donName == "" { + if len(h.donNames) != 1 { + return nil, fmt.Errorf("don_name is required: handler serves %d DONs", len(h.donNames)) + } + donName = h.donNames[0] } - return h.shards[idx], nil + var shardCount int + for _, shard := range h.shards { + if shard.DonName != donName { + continue + } + shardCount++ + if shard.ShardIdx == sel.ShardIndex { + return shard, nil + } + } + if shardCount == 0 { + return nil, fmt.Errorf("unknown DON %q", donName) + } + return nil, fmt.Errorf("shard index %d out of range: DON %q has %d shard(s)", sel.ShardIndex, donName, shardCount) } type activeRequest struct { @@ -231,16 +248,17 @@ type Config struct { type handler struct { services.StateMachine - // shards is the DON×shard matrix this handler relays for. Quorum is evaluated - // independently per shard (each with its own F and members); the first shard to - // reach quorum produces the user response. + // shards is the DON×shard matrix this handler relays for. Each request names one + // shard; quorum is evaluated against that shard's own F and members. shards []*gwhandlers.ShardEndpoint // donID names the handler for logging/metrics; it is the first shard's DON name. - donID string - codec api.JSONRPCCodec - lggr logger.Logger - mu sync.RWMutex - stopCh services.StopChan + donID string + // donNames lists the distinct DON names behind shards, in config order. + donNames []string + codec api.JSONRPCCodec + lggr logger.Logger + mu sync.RWMutex + stopCh services.StopChan globalNodeRateLimiter limits.RateLimiter perNodeRateLimiters map[string]limits.RateLimiter @@ -299,6 +317,12 @@ func NewHandler(methodConfig json.RawMessage, dons *gwhandlers.ShardedDONs, lggr return nil, errors.New("confidential relay handler requires at least one shard") } donID := shards[0].DonName + var donNames []string + for _, shard := range shards { + if !slices.Contains(donNames, shard.DonName) { + donNames = append(donNames, shard.DonName) + } + } globalNodeRateLimiter, err := limitsFactory.MakeRateLimiter(cresettings.Default.GatewayConfidentialRelayGlobalRate) if err != nil { @@ -323,6 +347,7 @@ func NewHandler(methodConfig json.RawMessage, dons *gwhandlers.ShardedDONs, lggr return &handler{ shards: shards, donID: donID, + donNames: donNames, lggr: logger.Named(lggr, "ConfidentialRelayHandler:"+donID), requestTimeout: time.Duration(cfg.RequestTimeoutSec) * time.Second, nodeSendTimeout: time.Duration(cfg.NodeSendTimeoutSec) * time.Second, @@ -433,11 +458,11 @@ func (h *handler) HandleJSONRPCUserMessage(ctx context.Context, req jsonrpc.Requ return errors.New("request ID is too long: " + strconv.Itoa(len(req.ID)) + ". max is 200 characters") } - shardIndex, err := extractShardIndex(req) + sel, err := extractShardSelector(req) if err != nil { return err } - shard, err := h.shardAt(shardIndex) + shard, err := h.resolveShard(sel) if err != nil { return err } diff --git a/core/services/gateway/handlers/confidentialrelay/handler_test.go b/core/services/gateway/handlers/confidentialrelay/handler_test.go index ba094477bfe..ffb7b342186 100644 --- a/core/services/gateway/handlers/confidentialrelay/handler_test.go +++ b/core/services/gateway/handlers/confidentialrelay/handler_test.go @@ -1403,5 +1403,72 @@ func TestConfidentialRelayHandler_ShardIndexOutOfRange(t *testing.T) { } err := h.HandleJSONRPCUserMessage(t.Context(), req, cb) - require.EqualError(t, err, "shard index 1 out of range: handler has 1 shard(s)") + require.EqualError(t, err, `shard index 1 out of range: DON "test_relay_don" has 1 shard(s)`) +} + +func setupTwoDONHandler(t *testing.T) (*handler, *common.Callback, []*mocks.DON) { + t.Helper() + lggr := logger.Test(t) + names := []string{"don_a", "don_b"} + cfgs := make([]config.ShardedDONConfig, 0, len(names)) + conns := make([][]gwhandlers.DON, 0, len(names)) + mockConns := make([]*mocks.DON, 0, len(names)) + addr := 0 + for _, name := range names { + members := make([]config.NodeConfig, 4) + for j := range members { + members[j] = config.NodeConfig{Name: fmt.Sprintf("node%d", addr), Address: fmt.Sprintf("0x%04d", addr)} + addr++ + } + don := mocks.NewDON(t) + mockConns = append(mockConns, don) + cfgs = append(cfgs, config.ShardedDONConfig{DonName: name, F: 1, Shards: []config.Shard{{Nodes: members}}}) + conns = append(conns, []gwhandlers.DON{don}) + } + dons, err := gwhandlers.NewShardedDONs(cfgs, conns) + require.NoError(t, err) + methodConfig, err := json.Marshal(Config{RequestTimeoutSec: 30}) + require.NoError(t, err) + h, err := NewHandler(methodConfig, dons, lggr, clockwork.NewFakeClock(), limits.Factory{Settings: cresettings.DefaultGetter, Logger: lggr}) + require.NoError(t, err) + return h, common.NewCallback(), mockConns +} + +func capParamsForDON(t *testing.T, donName string) json.RawMessage { + t.Helper() + params := validCapParams("wf1") + params.DonName = donName + raw, err := json.Marshal(params) + require.NoError(t, err) + return raw +} + +// don_name selects the DON: fan-out reaches only don_b's connection manager. +func TestConfidentialRelayHandler_DonNameSelectsDON(t *testing.T) { + t.Parallel() + h, cb, conns := setupTwoDONHandler(t) + conns[1].On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(nil) + + params := capParamsForDON(t, "don_b") + req := jsonrpc.Request[json.RawMessage]{ID: "req-don-b", Method: MethodCapabilityExec, Params: ¶ms} + require.NoError(t, h.HandleJSONRPCUserMessage(t.Context(), req, cb)) + require.Equal(t, "don_b", h.getActiveRequest(req.ID).shard.DonName) +} + +func TestConfidentialRelayHandler_DonNameRequiredWithMultipleDONs(t *testing.T) { + t.Parallel() + h, cb, _ := setupTwoDONHandler(t) + + params := capParamsForDON(t, "") + req := jsonrpc.Request[json.RawMessage]{ID: "req-no-don", Method: MethodCapabilityExec, Params: ¶ms} + require.EqualError(t, h.HandleJSONRPCUserMessage(t.Context(), req, cb), "don_name is required: handler serves 2 DONs") +} + +func TestConfidentialRelayHandler_UnknownDonName(t *testing.T) { + t.Parallel() + h, cb, _ := setupTwoDONHandler(t) + + params := capParamsForDON(t, "don_c") + req := jsonrpc.Request[json.RawMessage]{ID: "req-bad-don", Method: MethodCapabilityExec, Params: ¶ms} + require.EqualError(t, h.HandleJSONRPCUserMessage(t.Context(), req, cb), `unknown DON "don_c"`) } diff --git a/go.mod b/go.mod index f0bb2d12da8..8d5bb7ce360 100644 --- a/go.mod +++ b/go.mod @@ -82,7 +82,7 @@ require ( github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260908155618-50f521d70e62 github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260908155618-50f521d70e62 github.com/smartcontractkit/chainlink-ccv v0.14.0 - github.com/smartcontractkit/chainlink-common v0.11.2-0.20261015023017-0cf17e19699d + github.com/smartcontractkit/chainlink-common v0.11.2-0.20261016030140-129ade4252b6 github.com/smartcontractkit/chainlink-common/keystore v1.3.1-0.20260903141829-ef07b52a737d github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549 github.com/smartcontractkit/chainlink-data-streams v1.1.2-0.20261009064810-68424c714170 @@ -93,7 +93,7 @@ require ( github.com/smartcontractkit/chainlink-framework/chains v0.0.0-20260724153515-bb6a2de39bcb github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20260625152110-9afcf56e4053 github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4 - github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260916140808-f5d22b3cb3ee + github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20261013212015-e622e8b83b6b github.com/smartcontractkit/chainlink-protos/data-feeds v0.1.1-0.20260501174546-2e8846986b36 github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260729184203-90b4cdd48536 diff --git a/go.sum b/go.sum index 4ceec8c135b..2798197340a 100644 --- a/go.sum +++ b/go.sum @@ -1110,6 +1110,8 @@ github.com/smartcontractkit/chainlink-common v0.11.2-0.20261014212707-0e9847f396 github.com/smartcontractkit/chainlink-common v0.11.2-0.20261014212707-0e9847f3964e/go.mod h1:37Mu0pbAnGmHPgOZ8bQvgP4hzXw/Wu7smKX4cN46UOs= github.com/smartcontractkit/chainlink-common v0.11.2-0.20261015023017-0cf17e19699d h1:gwdUsOgsof08OhknvnEMM/PGfjahCOZNfhCDeffyfi4= github.com/smartcontractkit/chainlink-common v0.11.2-0.20261015023017-0cf17e19699d/go.mod h1:37Mu0pbAnGmHPgOZ8bQvgP4hzXw/Wu7smKX4cN46UOs= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20261016030140-129ade4252b6 h1:5J1qGIvA96iwGEDR0EXJFI6532Ozf7OxzaILnaUe52A= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20261016030140-129ade4252b6/go.mod h1:QeHe1mlcVzzg+awYAkRfl/8mDLRAhsl18IIOg4MVZec= github.com/smartcontractkit/chainlink-common/keystore v1.3.1-0.20260903141829-ef07b52a737d h1:M9xuiIvdiHtim4nlxCVajyC/Mk63tAg9ESWb/biAhNA= github.com/smartcontractkit/chainlink-common/keystore v1.3.1-0.20260903141829-ef07b52a737d/go.mod h1:rI2njmTlyTvh+xhmORmK6qX3+V1ycDAqGJLu8LS8WpU= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549 h1:J4vYRJxkBO8Yf//17wScKjYWmS45VZaFWx0zt+4ETEU= @@ -1146,6 +1148,8 @@ github.com/smartcontractkit/chainlink-protos/chainlink-ccv/verifier v0.0.0-20251 github.com/smartcontractkit/chainlink-protos/chainlink-ccv/verifier v0.0.0-20251211142334-5c3421fe2c8d/go.mod h1:5JdppgngCOUS76p61zCinSCgOhPeYQ+OcDUuome5THQ= github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260916140808-f5d22b3cb3ee h1:drewC9gPerS+FDeaaSfG7YL1Zzt/0Jr6YQ1YvHXzz04= github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260916140808-f5d22b3cb3ee/go.mod h1:/i8hjTPFdVWHiY+QjeSiVS2Z3GB3WAZznGgXHstC02E= +github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20261013212015-e622e8b83b6b h1:tlxPkVQ6ZpJIElqUWc4lU5VH4BB6OzhckG/gHwpaaa0= +github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20261013212015-e622e8b83b6b/go.mod h1:/i8hjTPFdVWHiY+QjeSiVS2Z3GB3WAZznGgXHstC02E= github.com/smartcontractkit/chainlink-protos/data-feeds v0.1.1-0.20260501174546-2e8846986b36 h1:SG+wAsNyAcA6Kk19ljuxi3HK9Ll2lpHik8OKoY4x7A0= github.com/smartcontractkit/chainlink-protos/data-feeds v0.1.1-0.20260501174546-2e8846986b36/go.mod h1:vL1bDgPSJjV0EqHYs4dDlR+EEE0cJchgvGLYXhwIjXY= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 h1:NJdGFhzT6zMaTod4QkBqVD2sg0I25iw1boOYtTpEwRo= From 5fb9f002e81b453aec4491eb4c188a0dd23d303d Mon Sep 17 00:00:00 2001 From: Bolek Kulbabinski <1416262+bolekk@users.noreply.github.com> Date: Thu, 15 Oct 2026 23:20:47 -0400 Subject: [PATCH 5/5] cre-6798: adapt relay shard selection to workflow_don_name/workflow_don_shard_index Follows the field rename in chainlink-common#2476. Co-Authored-By: Claude Sonnet 5.5 --- .../handlers/confidentialrelay/handler.go | 11 +++++---- .../confidentialrelay/handler_test.go | 24 +++++++++---------- go.mod | 2 +- go.sum | 2 ++ 4 files changed, 21 insertions(+), 18 deletions(-) diff --git a/core/services/gateway/handlers/confidentialrelay/handler.go b/core/services/gateway/handlers/confidentialrelay/handler.go index 0a9601b8e00..98fa4e50e10 100644 --- a/core/services/gateway/handlers/confidentialrelay/handler.go +++ b/core/services/gateway/handlers/confidentialrelay/handler.go @@ -107,11 +107,12 @@ func (h *handler) extractRequestLabels(req jsonrpc.Request[json.RawMessage]) req return labels } -// requestShardSelector is the routing fields both relay methods' params carry. A request -// that omits shard_index defaults to shard 0; one that omits don_name targets the only DON. +// requestShardSelector is the routing fields both relay methods' params carry: the +// workflow DON that started the execution. A request that omits the shard index defaults +// to shard 0; one that omits the DON name targets the only DON. type requestShardSelector struct { - DonName string `json:"don_name"` - ShardIndex int `json:"shard_index"` + DonName string `json:"workflow_don_name"` + ShardIndex int `json:"workflow_don_shard_index"` } func extractShardSelector(req jsonrpc.Request[json.RawMessage]) (requestShardSelector, error) { @@ -130,7 +131,7 @@ func (h *handler) resolveShard(sel requestShardSelector) (*gwhandlers.ShardEndpo donName := sel.DonName if donName == "" { if len(h.donNames) != 1 { - return nil, fmt.Errorf("don_name is required: handler serves %d DONs", len(h.donNames)) + return nil, fmt.Errorf("workflow_don_name is required: handler serves %d DONs", len(h.donNames)) } donName = h.donNames[0] } diff --git a/core/services/gateway/handlers/confidentialrelay/handler_test.go b/core/services/gateway/handlers/confidentialrelay/handler_test.go index ffb7b342186..348a93df148 100644 --- a/core/services/gateway/handlers/confidentialrelay/handler_test.go +++ b/core/services/gateway/handlers/confidentialrelay/handler_test.go @@ -883,7 +883,7 @@ func TestConfidentialRelayHandler_DuplicateRequestID(t *testing.T) { h, cb, don, _ := setupHandler(t, 4) don.On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(nil) - params := json.RawMessage(`{"workflow_id":"wf1","shard_index":0}`) + params := json.RawMessage(`{"workflow_id":"wf1"}`) req := jsonrpc.Request[json.RawMessage]{ ID: "req-dup", Method: MethodCapabilityExec, @@ -925,7 +925,7 @@ func TestConfidentialRelayHandler_RateLimitedNode(t *testing.T) { don.On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(nil) cb := common.NewCallback() - params := json.RawMessage(`{"workflow_id":"wf1","shard_index":0}`) + params := json.RawMessage(`{"workflow_id":"wf1"}`) req := jsonrpc.Request[json.RawMessage]{ ID: "req-ratelimit", Method: MethodCapabilityExec, @@ -1010,7 +1010,7 @@ func TestConfidentialRelayHandler_AllNodesFanOutFail(t *testing.T) { h, cb, don, _ := setupHandler(t, 4) don.On("SendToNode", mock.Anything, mock.Anything, mock.Anything).Return(errors.New("connection refused")) - params := json.RawMessage(`{"workflow_id":"wf1","shard_index":0}`) + params := json.RawMessage(`{"workflow_id":"wf1"}`) req := jsonrpc.Request[json.RawMessage]{ ID: "req-allfail", Method: MethodCapabilityExec, @@ -1051,7 +1051,7 @@ func TestConfidentialRelayHandler_FanOutWaitsWhileQuorumStillPossible(t *testing }, ) - params := json.RawMessage(`{"workflow_id":"wf1","shard_index":0}`) + params := json.RawMessage(`{"workflow_id":"wf1"}`) req := jsonrpc.Request[json.RawMessage]{ ID: "req-still-possible", Method: MethodCapabilityExec, @@ -1083,7 +1083,7 @@ func TestConfidentialRelayHandler_FanOutFailsWhenQuorumBecomesImpossible(t *test }, ) - params := json.RawMessage(`{"workflow_id":"wf1","shard_index":0}`) + params := json.RawMessage(`{"workflow_id":"wf1"}`) req := jsonrpc.Request[json.RawMessage]{ ID: "req-quorum-impossible", Method: MethodCapabilityExec, @@ -1135,7 +1135,7 @@ func TestConfidentialRelayHandler_FanOutToNodes_IsConcurrent(t *testing.T) { require.NoError(t, err) cb := common.NewCallback() - params := json.RawMessage(`{"workflow_id":"wf1","shard_index":0}`) + params := json.RawMessage(`{"workflow_id":"wf1"}`) req := jsonrpc.Request[json.RawMessage]{ ID: "req-concurrent-fanout", Method: MethodCapabilityExec, @@ -1220,7 +1220,7 @@ func TestConfidentialRelayHandler_BlockedNodeDoesNotStallFanOut(t *testing.T) { // Shortened so the test exercises the bound without waiting the production default. h.nodeSendTimeout = 50 * time.Millisecond - params := json.RawMessage(`{"workflow_id":"wf1","shard_index":0}`) + params := json.RawMessage(`{"workflow_id":"wf1"}`) req := jsonrpc.Request[json.RawMessage]{ ID: "req-blocked-node", Method: MethodCapabilityExec, @@ -1297,7 +1297,7 @@ func setupMultiShardHandler(t *testing.T, f int, shardSizes ...int) (*handler, * func capParamsForShard(t *testing.T, workflowID string, shardIndex int) json.RawMessage { t.Helper() params := validCapParams(workflowID) - params.ShardIndex = shardIndex + params.WorkflowDONShardIndex = shardIndex raw, err := json.Marshal(params) require.NoError(t, err) return raw @@ -1355,7 +1355,7 @@ func TestConfidentialRelayHandler_Sharded_RoutesToSelectedShardOnly(t *testing.T require.Nil(t, h.getActiveRequest(req.ID)) } -// A request that omits shard_index defaults to shard 0: fan-out reaches only shard 0's +// A request that omits workflow_don_shard_index defaults to shard 0: fan-out reaches only shard 0's // connection manager, never shard 1's. func TestConfidentialRelayHandler_ShardIndexMissingDefaultsToShardZero(t *testing.T) { t.Parallel() @@ -1437,13 +1437,13 @@ func setupTwoDONHandler(t *testing.T) (*handler, *common.Callback, []*mocks.DON) func capParamsForDON(t *testing.T, donName string) json.RawMessage { t.Helper() params := validCapParams("wf1") - params.DonName = donName + params.WorkflowDonName = donName raw, err := json.Marshal(params) require.NoError(t, err) return raw } -// don_name selects the DON: fan-out reaches only don_b's connection manager. +// workflow_don_name selects the DON: fan-out reaches only don_b's connection manager. func TestConfidentialRelayHandler_DonNameSelectsDON(t *testing.T) { t.Parallel() h, cb, conns := setupTwoDONHandler(t) @@ -1461,7 +1461,7 @@ func TestConfidentialRelayHandler_DonNameRequiredWithMultipleDONs(t *testing.T) params := capParamsForDON(t, "") req := jsonrpc.Request[json.RawMessage]{ID: "req-no-don", Method: MethodCapabilityExec, Params: ¶ms} - require.EqualError(t, h.HandleJSONRPCUserMessage(t.Context(), req, cb), "don_name is required: handler serves 2 DONs") + require.EqualError(t, h.HandleJSONRPCUserMessage(t.Context(), req, cb), "workflow_don_name is required: handler serves 2 DONs") } func TestConfidentialRelayHandler_UnknownDonName(t *testing.T) { diff --git a/go.mod b/go.mod index 8d5bb7ce360..69cadc72054 100644 --- a/go.mod +++ b/go.mod @@ -82,7 +82,7 @@ require ( github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260908155618-50f521d70e62 github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260908155618-50f521d70e62 github.com/smartcontractkit/chainlink-ccv v0.14.0 - github.com/smartcontractkit/chainlink-common v0.11.2-0.20261016030140-129ade4252b6 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20261016031554-0f929ebe3627 github.com/smartcontractkit/chainlink-common/keystore v1.3.1-0.20260903141829-ef07b52a737d github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549 github.com/smartcontractkit/chainlink-data-streams v1.1.2-0.20261009064810-68424c714170 diff --git a/go.sum b/go.sum index 2798197340a..a967552e240 100644 --- a/go.sum +++ b/go.sum @@ -1112,6 +1112,8 @@ github.com/smartcontractkit/chainlink-common v0.11.2-0.20261015023017-0cf17e1969 github.com/smartcontractkit/chainlink-common v0.11.2-0.20261015023017-0cf17e19699d/go.mod h1:37Mu0pbAnGmHPgOZ8bQvgP4hzXw/Wu7smKX4cN46UOs= github.com/smartcontractkit/chainlink-common v0.11.2-0.20261016030140-129ade4252b6 h1:5J1qGIvA96iwGEDR0EXJFI6532Ozf7OxzaILnaUe52A= github.com/smartcontractkit/chainlink-common v0.11.2-0.20261016030140-129ade4252b6/go.mod h1:QeHe1mlcVzzg+awYAkRfl/8mDLRAhsl18IIOg4MVZec= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20261016031554-0f929ebe3627 h1:PzWzuoSPl6Am2JeCOyJ9qFgcy0vKqrIIl3AXAXPE/2g= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20261016031554-0f929ebe3627/go.mod h1:QeHe1mlcVzzg+awYAkRfl/8mDLRAhsl18IIOg4MVZec= github.com/smartcontractkit/chainlink-common/keystore v1.3.1-0.20260903141829-ef07b52a737d h1:M9xuiIvdiHtim4nlxCVajyC/Mk63tAg9ESWb/biAhNA= github.com/smartcontractkit/chainlink-common/keystore v1.3.1-0.20260903141829-ef07b52a737d/go.mod h1:rI2njmTlyTvh+xhmORmK6qX3+V1ycDAqGJLu8LS8WpU= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549 h1:J4vYRJxkBO8Yf//17wScKjYWmS45VZaFWx0zt+4ETEU=