From e15d9e572165879bdfd22f73ee5849f5699ecc84 Mon Sep 17 00:00:00 2001 From: cawthorne Date: Wed, 7 Oct 2026 01:58:08 +0100 Subject: [PATCH 1/9] fix(chip-router): populate per-event PublishBatch results MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PublishBatch always returned an empty PublishResponse regardless of the batch's outcome. The node-side batch client (chainlink-common/pkg/chipingress/batch) requires one PublishResult per event to resolve delivery: since chainlink-common#2326 an empty results array is reported to every event's callback as ErrCodeResultsMismatch, so a chainlink v2.67 node's durable emitter against this router treated every event as undelivered and retransmitted the whole stream every 60s, forever — measured at 301k 'failed to deliver event. Relying on retransmit.' warnings on a single node within minutes of startup, a ~20x duplicated Kafka topic, and metric delivery-latency gaps long enough to break downstream freshness gates. - Ack each event (PublishResult{EventId}, nil error) when the batch was handed to at least one subscriber; the events were forwarded in that case, so the caller can resolve delivery and stop retransmitting. - Report a per-event PublishError when no subscriber accepted the batch, so the caller retains and retries (at-least-once preserved). - Return Unavailable when no subscribers are registered instead of an empty success — the previous behavior acked events that went nowhere, letting a durable store delete them undelivered (the data-loss shape flagged by the audit behind chainlink-common#2326). Bumps the module's chainlink-common/pkg/chipingress pin (Dec 2025 -> Sep 2026) — the old pin's pb predates PublishResult.Error. --- .../chiprouter/cmd/chip-router/main.go | 30 +++++- .../chiprouter/cmd/chip-router/main_test.go | 97 +++++++++++++++++++ framework/components/chiprouter/go.mod | 6 +- framework/components/chiprouter/go.sum | 8 +- 4 files changed, 132 insertions(+), 9 deletions(-) create mode 100644 framework/components/chiprouter/cmd/chip-router/main_test.go diff --git a/framework/components/chiprouter/cmd/chip-router/main.go b/framework/components/chiprouter/cmd/chip-router/main.go index 516d5b71f..2cb04dc61 100644 --- a/framework/components/chiprouter/cmd/chip-router/main.go +++ b/framework/components/chiprouter/cmd/chip-router/main.go @@ -11,6 +11,7 @@ import ( "os/signal" "strings" "sync" + "sync/atomic" "syscall" "time" @@ -19,7 +20,9 @@ import ( chippb "github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb" "golang.org/x/sync/errgroup" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" "github.com/smartcontractkit/chainlink-testing-framework/framework" ) @@ -164,9 +167,12 @@ func (r *router) Publish(ctx context.Context, event *cepb.CloudEvent) (*chippb.P func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch) (*chippb.PublishResponse, error) { snapshot := r.snapshotSubscribers() if len(snapshot) == 0 { - return &chippb.PublishResponse{}, nil + // Nothing will carry these events. Reporting an empty success here + // would let the caller's durable store delete them undelivered. + return nil, status.Error(codes.Unavailable, "no subscribers registered") } + var forwarded atomic.Bool var group errgroup.Group group.SetLimit(forwardParallel) for _, sub := range snapshot { @@ -177,13 +183,33 @@ func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch _, err := sub.client.PublishBatch(forwardCtx, batch) if err != nil { framework.L.Error().Msgf("chip router failed to forward batch to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err) + } else { + forwarded.Store(true) } framework.L.Debug().Msgf("chip router forwarded batch to subscriber id=%s", sub.id) return nil }) } _ = group.Wait() - return &chippb.PublishResponse{}, nil + + // PublishBatch callers (the node-side batch client driving the durable + // emitter) require one result per event to resolve delivery: an empty + // results array reads as RESULTS_MISMATCH and retransmits the whole + // stream forever. A batch the router handed to at least one subscriber + // counts as accepted for every event; a batch no subscriber accepted is + // reported failed per event so the caller retains and retries. + results := make([]*chippb.PublishResult, 0, len(batch.GetEvents())) + for _, ev := range batch.GetEvents() { + result := &chippb.PublishResult{EventId: ev.GetId()} + if !forwarded.Load() { + result.Error = &chippb.PublishError{ + ErrorCode: chippb.PublishErrorCode_PUBLISH_ERROR_CODE_UNKNOWN, + Reason: "chip router could not forward the batch to any subscriber", + } + } + results = append(results, result) + } + return &chippb.PublishResponse{Results: results}, nil } func (r *router) handleHealth(w http.ResponseWriter, req *http.Request) { diff --git a/framework/components/chiprouter/cmd/chip-router/main_test.go b/framework/components/chiprouter/cmd/chip-router/main_test.go new file mode 100644 index 000000000..25dc6957d --- /dev/null +++ b/framework/components/chiprouter/cmd/chip-router/main_test.go @@ -0,0 +1,97 @@ +package main + +import ( + "context" + "sync" + "testing" + + cepb "github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb" + chippb "github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type stubPublishClient struct { + mu sync.Mutex + batches int + err error +} + +func (s *stubPublishClient) Publish(context.Context, *cepb.CloudEvent, ...grpc.CallOption) (*chippb.PublishResponse, error) { + return &chippb.PublishResponse{}, nil +} + +func (s *stubPublishClient) PublishBatch(context.Context, *chippb.CloudEventBatch, ...grpc.CallOption) (*chippb.PublishResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.batches++ + return &chippb.PublishResponse{}, s.err +} + +func (s *stubPublishClient) batchCount() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.batches +} + +func batchOf(ids ...string) *chippb.CloudEventBatch { + events := make([]*cepb.CloudEvent, 0, len(ids)) + for _, id := range ids { + events = append(events, &cepb.CloudEvent{Id: id, Source: "s", SpecVersion: "1.0", Type: "t"}) + } + return &chippb.CloudEventBatch{Events: events} +} + +func TestPublishBatchPopulatesPerEventResults(t *testing.T) { + r := &router{subscribers: map[string]*subscriber{ + "sink": {id: "sink", client: &stubPublishClient{}}, + }} + + resp, err := r.PublishBatch(t.Context(), batchOf("e1", "e2", "e3")) + if err != nil { + t.Fatalf("PublishBatch: %v", err) + } + if len(resp.Results) != 3 { + t.Fatalf("want 3 results, got %d", len(resp.Results)) + } + for i, want := range []string{"e1", "e2", "e3"} { + if resp.Results[i].EventId != want { + t.Errorf("results[%d].EventId = %q, want %q", i, resp.Results[i].EventId, want) + } + if resp.Results[i].Error != nil { + t.Errorf("results[%d].Error = %v, want nil", i, resp.Results[i].Error) + } + } +} + +func TestPublishBatchNoSubscribersIsUnavailable(t *testing.T) { + r := &router{subscribers: map[string]*subscriber{}} + + _, err := r.PublishBatch(t.Context(), batchOf("e1")) + if err == nil { + t.Fatal("want error with no subscribers, got nil") + } + if status.Code(err) != codes.Unavailable { + t.Errorf("want Unavailable, got %v", status.Code(err)) + } +} + +func TestPublishBatchAllForwardsFailedReportsPerEventErrors(t *testing.T) { + r := &router{subscribers: map[string]*subscriber{ + "sink": {id: "sink", client: &stubPublishClient{err: context.DeadlineExceeded}}, + }} + + resp, err := r.PublishBatch(t.Context(), batchOf("e1", "e2")) + if err != nil { + t.Fatalf("PublishBatch: %v", err) + } + if len(resp.Results) != 2 { + t.Fatalf("want 2 results, got %d", len(resp.Results)) + } + for i, res := range resp.Results { + if res.Error == nil { + t.Errorf("results[%d].Error = nil, want a forward-failure error", i) + } + } +} diff --git a/framework/components/chiprouter/go.mod b/framework/components/chiprouter/go.mod index c31193136..18347708e 100644 --- a/framework/components/chiprouter/go.mod +++ b/framework/components/chiprouter/go.mod @@ -1,6 +1,6 @@ module github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter -go 1.26.5 +go 1.26.6 replace github.com/smartcontractkit/chainlink-testing-framework/framework => ../../ @@ -9,11 +9,11 @@ require ( github.com/google/uuid v1.6.0 github.com/moby/moby/api v1.54.1 github.com/pkg/errors v0.9.1 - github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20251211140724-319861e514c4 + github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260915184316-2730f1867c92 github.com/smartcontractkit/chainlink-testing-framework/framework v0.15.8 github.com/testcontainers/testcontainers-go v0.42.0 golang.org/x/sync v0.22.0 - google.golang.org/grpc v1.83.1 + google.golang.org/grpc v1.83.2 ) require ( diff --git a/framework/components/chiprouter/go.sum b/framework/components/chiprouter/go.sum index 52871d8a1..38bf00c87 100644 --- a/framework/components/chiprouter/go.sum +++ b/framework/components/chiprouter/go.sum @@ -205,8 +205,8 @@ github.com/shirou/gopsutil/v4 v4.26.3 h1:2ESdQt90yU3oXF/CdOlRCJxrP+Am1aBYubTMTfx github.com/shirou/gopsutil/v4 v4.26.3/go.mod h1:LZ6ewCSkBqUpvSOf+LsTGnRinC6iaNUNMGBtDkJBaLQ= github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20251211140724-319861e514c4 h1:NOUsjsMzNecbjiPWUQGlRSRAutEvCFrqqyETDJeh5q4= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20251211140724-319861e514c4/go.mod h1:Zpvul9sTcZNAZOVzt5vBl1XZGNvQebFpnpn3/KOQvOQ= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260915184316-2730f1867c92 h1:Z1OIGukDhJD+Gknr5G07Vtml24E7BwKaIQoWzV/SdqA= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260915184316-2730f1867c92/go.mod h1:Scyk7O80fde5rJF3Ty3fEE3xiS6j84eWbu8/2gl3pIY= github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o= github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= @@ -278,8 +278,8 @@ gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= google.golang.org/genproto/googleapis/rpc v0.0.0-20260819154853-08b0e4226688 h1:cYNAzI2sUwhmCcoj9TxvihSrqsxt6uIkj3rDRhSDmW4= google.golang.org/genproto/googleapis/rpc v0.0.0-20260819154853-08b0e4226688/go.mod h1:DjtHYE8FKJLivXcBEjGwndXfIC23G0VpXiXKqG179uA= -google.golang.org/grpc v1.83.1 h1:HIO0+BEtBP6soyqvqC8sNUjZ7bTs+0hFQuFF+RAy++Y= -google.golang.org/grpc v1.83.1/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ= +google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU= +google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsokw1S8= google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/evanphx/json-patch.v4 v4.12.0 h1:n6jtcsulIzXPJaxegRbvFNNrZDjbij7ny3gmSPG+6V4= From 066433d0105d8ebb7ccd2a5fd5d49ad6ee8924b0 Mon Sep 17 00:00:00 2001 From: cawthorne Date: Wed, 7 Oct 2026 10:51:12 +0100 Subject: [PATCH 2/9] fix(chip-router): aggregate downstream per-event outcomes; fail the RPC when nothing accepts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the review findings: - A nil downstream RPC error does not mean every event was accepted — the protocol supports per-event rejections in a successful response. The router now aggregates the subscribers' per-event results and passes a rejection through only while no subscriber has accepted the event. Subscribers that accept without per-event detail (nil-error response, empty results array) count as accepting the whole batch — the router trusts a subscriber's RPC success at the forwarding boundary, the same as Publish does. - When no subscriber accepts the batch, return Unavailable instead of a nil-error response with per-event errors: transactional callers resolve the whole batch from the RPC outcome alone, so per-event errors would still acknowledge everything. An RPC error makes every caller mode retain and retry. --- .../chiprouter/cmd/chip-router/main.go | 63 +++++++++++---- .../chiprouter/cmd/chip-router/main_test.go | 78 ++++++++++++++++--- 2 files changed, 115 insertions(+), 26 deletions(-) diff --git a/framework/components/chiprouter/cmd/chip-router/main.go b/framework/components/chiprouter/cmd/chip-router/main.go index 2cb04dc61..be6772163 100644 --- a/framework/components/chiprouter/cmd/chip-router/main.go +++ b/framework/components/chiprouter/cmd/chip-router/main.go @@ -11,7 +11,6 @@ import ( "os/signal" "strings" "sync" - "sync/atomic" "syscall" "time" @@ -172,7 +171,19 @@ func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch return nil, status.Error(codes.Unavailable, "no subscribers registered") } - var forwarded atomic.Bool + // Aggregate the downstream outcomes so the per-event results reflect what + // actually happened to each event: + // - anyAccepted: at least one subscriber's PublishBatch RPC succeeded. + // - detail: merged per-event results from the subscribers that provide + // them. An event is reported rejected only if every detail-providing + // subscriber rejected it; subscribers that accept without per-event + // detail (a nil-error response with an empty results array) count as + // accepting the whole batch — the router trusts a subscriber's RPC + // success at the forwarding boundary, the same as Publish does. + var mu sync.Mutex + anyAccepted := false + var detail []*chippb.PublishResult + var group errgroup.Group group.SetLimit(forwardParallel) for _, sub := range snapshot { @@ -180,32 +191,50 @@ func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch framework.L.Debug().Msgf("chip router forwarding batch to subscriber id=%s name=%s endpoint=%s", sub.id, sub.name, sub.endpoint) forwardCtx, cancel := context.WithTimeout(ctx, forwardTimeout) defer cancel() - _, err := sub.client.PublishBatch(forwardCtx, batch) + resp, err := sub.client.PublishBatch(forwardCtx, batch) if err != nil { framework.L.Error().Msgf("chip router failed to forward batch to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err) - } else { - forwarded.Store(true) + return nil } framework.L.Debug().Msgf("chip router forwarded batch to subscriber id=%s", sub.id) + mu.Lock() + defer mu.Unlock() + anyAccepted = true + if results := resp.GetResults(); len(results) > 0 { + if detail == nil { + detail = results + return nil + } + // Merge: an accepting result overrides a rejecting one for the + // same event index; a rejection sticks only while no subscriber + // has accepted the event. + for i := range detail { + if i >= len(results) { + break + } + if detail[i].GetError() != nil && results[i].GetError() == nil { + detail[i] = results[i] + } + } + } return nil }) } _ = group.Wait() - // PublishBatch callers (the node-side batch client driving the durable - // emitter) require one result per event to resolve delivery: an empty - // results array reads as RESULTS_MISMATCH and retransmits the whole - // stream forever. A batch the router handed to at least one subscriber - // counts as accepted for every event; a batch no subscriber accepted is - // reported failed per event so the caller retains and retries. + // A nil RPC error does not imply the batch was accepted: every forward + // can fail (or a transactional caller resolves the whole batch from the + // RPC outcome alone), so fail the RPC and let the caller retain and + // retry instead of acknowledging with per-event errors. + if !anyAccepted { + return nil, status.Error(codes.Unavailable, fmt.Sprintf("all %d subscribers failed to accept the batch", len(snapshot))) + } + results := make([]*chippb.PublishResult, 0, len(batch.GetEvents())) - for _, ev := range batch.GetEvents() { + for i, ev := range batch.GetEvents() { result := &chippb.PublishResult{EventId: ev.GetId()} - if !forwarded.Load() { - result.Error = &chippb.PublishError{ - ErrorCode: chippb.PublishErrorCode_PUBLISH_ERROR_CODE_UNKNOWN, - Reason: "chip router could not forward the batch to any subscriber", - } + if i < len(detail) && detail[i].GetError() != nil { + result.Error = detail[i].GetError() } results = append(results, result) } diff --git a/framework/components/chiprouter/cmd/chip-router/main_test.go b/framework/components/chiprouter/cmd/chip-router/main_test.go index 25dc6957d..19e59ecbb 100644 --- a/framework/components/chiprouter/cmd/chip-router/main_test.go +++ b/framework/components/chiprouter/cmd/chip-router/main_test.go @@ -13,9 +13,10 @@ import ( ) type stubPublishClient struct { - mu sync.Mutex - batches int - err error + mu sync.Mutex + batches int + err error + response *chippb.PublishResponse } func (s *stubPublishClient) Publish(context.Context, *cepb.CloudEvent, ...grpc.CallOption) (*chippb.PublishResponse, error) { @@ -26,7 +27,13 @@ func (s *stubPublishClient) PublishBatch(context.Context, *chippb.CloudEventBatc s.mu.Lock() defer s.mu.Unlock() s.batches++ - return &chippb.PublishResponse{}, s.err + if s.err != nil { + return nil, s.err + } + if s.response != nil { + return s.response, nil + } + return &chippb.PublishResponse{}, nil } func (s *stubPublishClient) batchCount() int { @@ -77,11 +84,38 @@ func TestPublishBatchNoSubscribersIsUnavailable(t *testing.T) { } } -func TestPublishBatchAllForwardsFailedReportsPerEventErrors(t *testing.T) { +func TestPublishBatchAllForwardsFailedIsUnavailable(t *testing.T) { r := &router{subscribers: map[string]*subscriber{ "sink": {id: "sink", client: &stubPublishClient{err: context.DeadlineExceeded}}, }} + // A nil RPC error with per-event errors would still acknowledge the whole + // batch for transactional callers (they resolve delivery from the RPC + // outcome alone), so the router must fail the RPC instead. + _, err := r.PublishBatch(t.Context(), batchOf("e1", "e2")) + if err == nil { + t.Fatal("want error when every forward fails, got nil") + } + if status.Code(err) != codes.Unavailable { + t.Errorf("want Unavailable, got %v", status.Code(err)) + } +} + +func TestPublishBatchAggregatesDownstreamRejections(t *testing.T) { + // One subscriber rejects e2 per-event (nil RPC error); the router must + // pass that rejection through rather than acknowledging the whole batch. + r := &router{subscribers: map[string]*subscriber{ + "sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{ + {EventId: "e1"}, + {EventId: "e2", Error: &chippb.PublishError{ + ErrorCode: chippb.PublishErrorCode_PUBLISH_ERROR_CODE_VALIDATION_FAILED, + Reason: "invalid event", + }}, + }, + }}}, + }} + resp, err := r.PublishBatch(t.Context(), batchOf("e1", "e2")) if err != nil { t.Fatalf("PublishBatch: %v", err) @@ -89,9 +123,35 @@ func TestPublishBatchAllForwardsFailedReportsPerEventErrors(t *testing.T) { if len(resp.Results) != 2 { t.Fatalf("want 2 results, got %d", len(resp.Results)) } - for i, res := range resp.Results { - if res.Error == nil { - t.Errorf("results[%d].Error = nil, want a forward-failure error", i) - } + if resp.Results[0].Error != nil { + t.Errorf("results[0].Error = %v, want nil (accepted)", resp.Results[0].Error) + } + if resp.Results[1].Error == nil { + t.Error("results[1].Error = nil, want the downstream rejection passed through") + } +} + +func TestPublishBatchAnAcceptingSubscriberOverridesADetaillessRejection(t *testing.T) { + // Two detail-providing subscribers disagree: the event is rejected by one + // and accepted by the other — acceptance wins. + r := &router{subscribers: map[string]*subscriber{ + "a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{ + {EventId: "e1", Error: &chippb.PublishError{Reason: "rejected by a"}}, + }, + }}}, + "b": {id: "b", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{ + {EventId: "e1"}, + }, + }}}, + }} + + resp, err := r.PublishBatch(t.Context(), batchOf("e1")) + if err != nil { + t.Fatalf("PublishBatch: %v", err) + } + if len(resp.Results) != 1 || resp.Results[0].Error != nil { + t.Errorf("want e1 accepted, got results=%+v", resp.Results) } } From e8fb57235b7d12fe0323f5faaf801014d09bcb35 Mon Sep 17 00:00:00 2001 From: cawthorne Date: Wed, 7 Oct 2026 11:04:43 +0100 Subject: [PATCH 3/9] chore: bump the chip_router example's go directive to 1.26.6 (gomods tidy) --- framework/examples/chip_router/go.mod | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/framework/examples/chip_router/go.mod b/framework/examples/chip_router/go.mod index 8a652b397..474e75575 100644 --- a/framework/examples/chip_router/go.mod +++ b/framework/examples/chip_router/go.mod @@ -1,6 +1,6 @@ module github.com/smartcontractkit/chainlink-testing-framework/framework/examples/chiprouter -go 1.26.5 +go 1.26.6 replace github.com/smartcontractkit/chainlink-testing-framework/framework => ../../ From 3c74ee1f968170f17a6410dc17e4f2195d42cfb8 Mon Sep 17 00:00:00 2001 From: cawthorne Date: Wed, 7 Oct 2026 11:29:53 +0100 Subject: [PATCH 4/9] docs: trim comments --- .../chiprouter/cmd/chip-router/main.go | 26 ++++++------------- .../chiprouter/cmd/chip-router/main_test.go | 4 +-- 2 files changed, 9 insertions(+), 21 deletions(-) diff --git a/framework/components/chiprouter/cmd/chip-router/main.go b/framework/components/chiprouter/cmd/chip-router/main.go index be6772163..b02f92e2b 100644 --- a/framework/components/chiprouter/cmd/chip-router/main.go +++ b/framework/components/chiprouter/cmd/chip-router/main.go @@ -166,20 +166,14 @@ func (r *router) Publish(ctx context.Context, event *cepb.CloudEvent) (*chippb.P func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch) (*chippb.PublishResponse, error) { snapshot := r.snapshotSubscribers() if len(snapshot) == 0 { - // Nothing will carry these events. Reporting an empty success here - // would let the caller's durable store delete them undelivered. + // Acking events nothing carried would let a durable store delete them undelivered. return nil, status.Error(codes.Unavailable, "no subscribers registered") } - // Aggregate the downstream outcomes so the per-event results reflect what - // actually happened to each event: - // - anyAccepted: at least one subscriber's PublishBatch RPC succeeded. - // - detail: merged per-event results from the subscribers that provide - // them. An event is reported rejected only if every detail-providing - // subscriber rejected it; subscribers that accept without per-event - // detail (a nil-error response with an empty results array) count as - // accepting the whole batch — the router trusts a subscriber's RPC - // success at the forwarding boundary, the same as Publish does. + // anyAccepted: at least one subscriber's RPC succeeded. detail: merged + // per-event results — an event is rejected only if every detail-providing + // subscriber rejected it; a subscriber accepting without per-event detail + // counts as accepting the whole batch. var mu sync.Mutex anyAccepted := false var detail []*chippb.PublishResult @@ -205,9 +199,7 @@ func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch detail = results return nil } - // Merge: an accepting result overrides a rejecting one for the - // same event index; a rejection sticks only while no subscriber - // has accepted the event. + // An accepting result overrides a rejection for the same event. for i := range detail { if i >= len(results) { break @@ -222,10 +214,8 @@ func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch } _ = group.Wait() - // A nil RPC error does not imply the batch was accepted: every forward - // can fail (or a transactional caller resolves the whole batch from the - // RPC outcome alone), so fail the RPC and let the caller retain and - // retry instead of acknowledging with per-event errors. + // Fail the RPC when nothing accepted: transactional callers resolve the + // whole batch from the RPC outcome alone. if !anyAccepted { return nil, status.Error(codes.Unavailable, fmt.Sprintf("all %d subscribers failed to accept the batch", len(snapshot))) } diff --git a/framework/components/chiprouter/cmd/chip-router/main_test.go b/framework/components/chiprouter/cmd/chip-router/main_test.go index 19e59ecbb..69bf248e8 100644 --- a/framework/components/chiprouter/cmd/chip-router/main_test.go +++ b/framework/components/chiprouter/cmd/chip-router/main_test.go @@ -89,9 +89,7 @@ func TestPublishBatchAllForwardsFailedIsUnavailable(t *testing.T) { "sink": {id: "sink", client: &stubPublishClient{err: context.DeadlineExceeded}}, }} - // A nil RPC error with per-event errors would still acknowledge the whole - // batch for transactional callers (they resolve delivery from the RPC - // outcome alone), so the router must fail the RPC instead. + // Transactional callers resolve the whole batch from the RPC outcome alone. _, err := r.PublishBatch(t.Context(), batchOf("e1", "e2")) if err == nil { t.Fatal("want error when every forward fails, got nil") From c69acae8c2faa2c6a8f33d22eb1d760d910c84d3 Mon Sep 17 00:00:00 2001 From: cawthorne Date: Wed, 7 Oct 2026 11:34:02 +0100 Subject: [PATCH 5/9] fix(chip-router): honor transaction_enabled; a downstream all-rejection fails the RPC MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A subscriber's nil RPC error does not mean its events were accepted: the pinned protocol reports per-event rejections in a successful response. Renamed the flag to anyForwarded (what an RPC success actually tells the router) and gated the response on the batch's own PublishOptions: - transaction_enabled: any unaccepted event fails the RPC (Unavailable) — transactional callers resolve the whole batch from the RPC outcome alone, so per-event errors would still acknowledge everything, including events no subscriber produced. - otherwise: per-event results as before (rejections passed through, acceptance wins over rejection across subscribers). --- .../chiprouter/cmd/chip-router/main.go | 22 ++++-- .../chiprouter/cmd/chip-router/main_test.go | 71 +++++++++++++++++++ 2 files changed, 87 insertions(+), 6 deletions(-) diff --git a/framework/components/chiprouter/cmd/chip-router/main.go b/framework/components/chiprouter/cmd/chip-router/main.go index b02f92e2b..e680b6a86 100644 --- a/framework/components/chiprouter/cmd/chip-router/main.go +++ b/framework/components/chiprouter/cmd/chip-router/main.go @@ -170,12 +170,12 @@ func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch return nil, status.Error(codes.Unavailable, "no subscribers registered") } - // anyAccepted: at least one subscriber's RPC succeeded. detail: merged + // anyForwarded: at least one subscriber's RPC succeeded. detail: merged // per-event results — an event is rejected only if every detail-providing // subscriber rejected it; a subscriber accepting without per-event detail // counts as accepting the whole batch. var mu sync.Mutex - anyAccepted := false + anyForwarded := false var detail []*chippb.PublishResult var group errgroup.Group @@ -193,7 +193,7 @@ func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch framework.L.Debug().Msgf("chip router forwarded batch to subscriber id=%s", sub.id) mu.Lock() defer mu.Unlock() - anyAccepted = true + anyForwarded = true if results := resp.GetResults(); len(results) > 0 { if detail == nil { detail = results @@ -214,9 +214,9 @@ func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch } _ = group.Wait() - // Fail the RPC when nothing accepted: transactional callers resolve the - // whole batch from the RPC outcome alone. - if !anyAccepted { + // No subscriber accepted the batch: fail the RPC so every caller mode + // retains and retries. + if !anyForwarded { return nil, status.Error(codes.Unavailable, fmt.Sprintf("all %d subscribers failed to accept the batch", len(snapshot))) } @@ -228,6 +228,16 @@ func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch } results = append(results, result) } + + // Transactional callers resolve the whole batch from the RPC outcome + // alone, so any unaccepted event must fail the RPC (all-or-nothing). + if batch.GetOptions().GetTransactionEnabled() { + for _, result := range results { + if result.Error != nil { + return nil, status.Error(codes.Unavailable, "not all events were accepted by a subscriber") + } + } + } return &chippb.PublishResponse{Results: results}, nil } diff --git a/framework/components/chiprouter/cmd/chip-router/main_test.go b/framework/components/chiprouter/cmd/chip-router/main_test.go index 69bf248e8..3ce630356 100644 --- a/framework/components/chiprouter/cmd/chip-router/main_test.go +++ b/framework/components/chiprouter/cmd/chip-router/main_test.go @@ -42,6 +42,8 @@ func (s *stubPublishClient) batchCount() int { return s.batches } +func ptr(b bool) *bool { return &b } + func batchOf(ids ...string) *chippb.CloudEventBatch { events := make([]*cepb.CloudEvent, 0, len(ids)) for _, id := range ids { @@ -129,6 +131,75 @@ func TestPublishBatchAggregatesDownstreamRejections(t *testing.T) { } } +func TestPublishBatchTransactionalPartialRejectionFailsRPC(t *testing.T) { + // Transactional callers resolve the whole batch from the RPC outcome + // alone, so a partial rejection must fail the RPC, not ack everything. + r := &router{subscribers: map[string]*subscriber{ + "sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{ + {EventId: "e1"}, + {EventId: "e2", Error: &chippb.PublishError{Reason: "rejected"}}, + }, + }}}, + }} + + batch := batchOf("e1", "e2") + batch.Options = &chippb.PublishOptions{TransactionEnabled: ptr(true)} + _, err := r.PublishBatch(t.Context(), batch) + if err == nil { + t.Fatal("want error for a partial rejection under transaction_enabled, got nil") + } + if status.Code(err) != codes.Unavailable { + t.Errorf("want Unavailable, got %v", status.Code(err)) + } +} + +func TestPublishBatchTransactionalAllRejectedByDownstreamFailsRPC(t *testing.T) { + // A nil downstream RPC error with every event rejected must NOT count as + // accepted: a transactional caller would otherwise delete them undelivered. + r := &router{subscribers: map[string]*subscriber{ + "sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{ + {EventId: "e1", Error: &chippb.PublishError{Reason: "rejected"}}, + {EventId: "e2", Error: &chippb.PublishError{Reason: "rejected"}}, + }, + }}}, + }} + + batch := batchOf("e1", "e2") + batch.Options = &chippb.PublishOptions{TransactionEnabled: ptr(true)} + _, err := r.PublishBatch(t.Context(), batch) + if err == nil { + t.Fatal("want error when the subscriber rejected every event, got nil") + } + if status.Code(err) != codes.Unavailable { + t.Errorf("want Unavailable, got %v", status.Code(err)) + } +} + +func TestPublishBatchTransactionalAllAcceptedPasses(t *testing.T) { + r := &router{subscribers: map[string]*subscriber{ + "sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{ + {EventId: "e1"}, + {EventId: "e2"}, + }, + }}}, + }} + + batch := batchOf("e1", "e2") + batch.Options = &chippb.PublishOptions{TransactionEnabled: ptr(true)} + resp, err := r.PublishBatch(t.Context(), batch) + if err != nil { + t.Fatalf("PublishBatch: %v", err) + } + for i, res := range resp.Results { + if res.Error != nil { + t.Errorf("results[%d].Error = %v, want nil", i, res.Error) + } + } +} + func TestPublishBatchAnAcceptingSubscriberOverridesADetaillessRejection(t *testing.T) { // Two detail-providing subscribers disagree: the event is rejected by one // and accepted by the other — acceptance wins. From 96b5d7e42e50c3620bc40d59177c6f90e72c5031 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Mon, 26 Oct 2026 08:50:02 -0700 Subject: [PATCH 6/9] fix(chip-router): match atlas write-service Publish/PublishBatch semantics (#2873) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(chip-router): match atlas write-service Publish/PublishBatch semantics Aligns the router with the atlas chip-ingress write-service, addressing the review findings on #2861: - Publish (single event) now follows the write-service's all-or-nothing contract (atlas chip-ingress internal/server: Publish delegates to PublishCloudEvents with transaction_enabled=true): it fails the RPC when no subscribers are registered, when every forward fails, or when no subscriber accepts the event, and returns exactly one PublishResult on success. Previously it returned an empty PublishResponse with a nil error unconditionally, acking events that went nowhere. - Failure RPCs use codes.Internal, the write-service's failure code, in place of codes.Unavailable. - An empty batch short-circuits to a no-op success before the subscriber check, matching the write-service's empty-batch path. - The per-event detail merge is now order-independent and sized to the batch: a later subscriber's results can extend detail beyond the first provider's length (previously trailing events were acked with no detail), and a detail-less success (empty results, nil error) now accepts the whole batch, clearing other subscribers' rejections — the intent documented in #2861 but not enforced by its merge. Tests: single-event success/failure paths, empty batch, detail-less acceptance, merge extension past the first provider's detail, rejection surviving another subscriber's RPC failure; stub simplified to an immutable struct. * fix(chip-router): bump builder image to golang:1.26.6 to match go.mod go.mod requires go >= 1.26.6 (bumped alongside the pkg/chipingress pin), so the pinned golang:1.25.3 builder fails at 'go mod download' with 'go: go.mod requires go >= 1.26.6 (running go 1.25.3; GOTOOLCHAIN=local)'. * fix(chip-router): implement Ping for write-service surface parity The write-service answers Ping with a static pong; the router inherited UnimplementedChipIngressServer's Unimplemented for it, which the new atlas parity suite's P0 surface probe flags as drift. Ping never needs to reach subscribers, so answer directly like the write-service. * fix(chip-router): make the image build self-contained (no framework dep in the binary) The image build recipe (chainlink cre environment: docker_ctx = framework/components/chiprouter) passes the module dir as the build context, so go.mod's `replace framework => ../../` (a monorepo-dev convenience inherited from main) cannot resolve inside the container and 'go mod download' fails. Published framework releases also do not compile against this module's moby pins, so the replace cannot simply be dropped for the module as a whole. The router binary only used framework for its global logger, so: - cmd/chip-router now carries its own zerolog logger, mirroring framework.L's setup (CTF_LOG_LEVEL env var, console writer to stderr); output format is unchanged. - the Dockerfile drops the replace for the isolated build and runs 'go mod tidy' so the published framework — needed only by the deploy-side library, which is not compiled into the image — resolves from the proxy for module-graph completeness. The deploy-side library (chiprouter.go) keeps its framework dependency and the monorepo replace. * fix(chip-router): reject mismatched batch results and tighten Publish failure path PublishBatch now treats a non-empty subscriber response whose result count differs from the batch size as a contract violation (the write-service returns exactly one result per event in input order): it is logged at Warn and counted as a failed forward instead of silently acking the events it does not cover. Empty responses stay a detail-less success, which chainlink's chip-testsink subscriber relies on. Publish drops the redundant anyForwarded flag and surfaces the subscriber's rejection reason in the Internal error when no subscriber accepted the event. Tests: rewrite the merge test into mismatch cases, make the detail-less acceptance test genuinely detail-less, and cover the single-subscriber empty-response Publish path. Also fix gofmt import order and make the package logger unexported. --- framework/components/chiprouter/Dockerfile | 14 +- .../chiprouter/cmd/chip-router/main.go | 155 ++++++--- .../chiprouter/cmd/chip-router/main_test.go | 323 ++++++++++++++---- framework/components/chiprouter/go.mod | 2 +- 4 files changed, 383 insertions(+), 111 deletions(-) diff --git a/framework/components/chiprouter/Dockerfile b/framework/components/chiprouter/Dockerfile index 8d998ae0a..af44ef24b 100644 --- a/framework/components/chiprouter/Dockerfile +++ b/framework/components/chiprouter/Dockerfile @@ -1,12 +1,18 @@ -FROM golang:1.25.3 AS builder +FROM golang:1.26.6 AS builder WORKDIR /src -COPY go.mod go.sum ./ -RUN go mod download - COPY . . +# The module's go.mod replaces the parent framework module with ../../ (a +# monorepo-dev convenience) which cannot resolve inside this build's +# module-dir context, and the published framework does not compile against +# this module's moby pins. The router binary no longer imports framework +# (only the deploy-side library does, and it is not compiled into the +# image), so drop the replace and fetch the published framework module from +# the proxy for module-graph completeness only. +RUN sed -i '/^replace /d' go.mod && go mod tidy && go mod download + ARG TARGETOS=linux ARG TARGETARCH=amd64 ARG CTF_LOG_LEVEL=info diff --git a/framework/components/chiprouter/cmd/chip-router/main.go b/framework/components/chiprouter/cmd/chip-router/main.go index e680b6a86..de6a7df73 100644 --- a/framework/components/chiprouter/cmd/chip-router/main.go +++ b/framework/components/chiprouter/cmd/chip-router/main.go @@ -16,16 +16,35 @@ import ( cepb "github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb" "github.com/google/uuid" + "github.com/rs/zerolog" + "github.com/rs/zerolog/log" chippb "github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb" "golang.org/x/sync/errgroup" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" - - "github.com/smartcontractkit/chainlink-testing-framework/framework" ) +// logger is the router's logger. It mirrors framework.L's setup (CTF_LOG_LEVEL +// env var, console writer to stderr) so output is unchanged now that the +// binary no longer imports the framework module — the image build must stay +// self-contained (see the Dockerfile: the monorepo-local `replace` cannot +// resolve in its module-dir context). +var logger zerolog.Logger + +func init() { + lvlStr := os.Getenv("CTF_LOG_LEVEL") + if lvlStr == "" { + lvlStr = "info" + } + lvl, err := zerolog.ParseLevel(lvlStr) + if err != nil { + panic(err) + } + logger = log.Output(zerolog.ConsoleWriter{Out: os.Stderr}).Level(lvl) +} + const ( defaultGRPCAddr = "0.0.0.0:50051" defaultAdminAddr = "0.0.0.0:50050" @@ -67,7 +86,7 @@ type healthResponse struct { func main() { if err := run(); err != nil { - framework.L.Error().Msgf("chip router failed: %v", err) + logger.Error().Msgf("chip router failed: %v", err) os.Exit(1) } } @@ -110,7 +129,7 @@ func run() error { go func() { errCh <- grpcServer.Serve(grpcLis) }() go func() { errCh <- adminServer.Serve(adminLis) }() - framework.L.Info().Msgf("chip router started: grpc=%s admin=%s", grpcAddr, adminAddr) + logger.Info().Msgf("chip router started: grpc=%s admin=%s", grpcAddr, adminAddr) select { case <-ctx.Done(): @@ -133,80 +152,128 @@ func run() error { if err == nil || errors.Is(err, http.ErrServerClosed) { return nil } - framework.L.Error().Msgf("chip router server error: %v", err) + logger.Error().Msgf("chip router server error: %v", err) return err } } +// Ping matches the write-service's Ping: a static health response that never +// reaches the subscribers. +func (r *router) Ping(context.Context, *chippb.EmptyRequest) (*chippb.PingResponse, error) { + return &chippb.PingResponse{Message: "pong"}, nil +} + +// Publish matches the write-service's single-event contract (atlas +// chip-ingress internal/server): the RPC is all-or-nothing — it fails unless +// at least one subscriber accepted the event, and returns exactly one +// PublishResult on success. func (r *router) Publish(ctx context.Context, event *cepb.CloudEvent) (*chippb.PublishResponse, error) { snapshot := r.snapshotSubscribers() if len(snapshot) == 0 { - return &chippb.PublishResponse{}, nil + // Acking an event nothing carried would let a durable store delete it undelivered. + return nil, status.Error(codes.Internal, "no subscribers registered") } + // accepted: some subscriber accepted the event (an accepted result or a + // detail-less success) — sticky, so a later rejection cannot unset it. + // rejection: recorded only while no subscriber has accepted. + var mu sync.Mutex + accepted := false + var rejection *chippb.PublishError + var group errgroup.Group group.SetLimit(forwardParallel) for _, sub := range snapshot { group.Go(func() error { - framework.L.Debug().Msgf("chip router forwarding event to subscriber id=%s name=%s endpoint=%s", sub.id, sub.name, sub.endpoint) + logger.Debug().Msgf("chip router forwarding event to subscriber id=%s name=%s endpoint=%s", sub.id, sub.name, sub.endpoint) forwardCtx, cancel := context.WithTimeout(ctx, forwardTimeout) defer cancel() - _, err := sub.client.Publish(forwardCtx, event) + resp, err := sub.client.Publish(forwardCtx, event) if err != nil { - framework.L.Error().Msgf("chip router failed to forward event to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err) + logger.Error().Msgf("chip router failed to forward event to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err) + return nil + } + logger.Debug().Msgf("chip router forwarded event to subscriber id=%s", sub.id) + mu.Lock() + defer mu.Unlock() + if results := resp.GetResults(); len(results) > 0 && results[0].GetError() != nil { + if !accepted && rejection == nil { + rejection = results[0].GetError() + } + return nil } - framework.L.Debug().Msgf("chip router forwarded event to subscriber id=%s", sub.id) + accepted = true return nil }) } _ = group.Wait() - return &chippb.PublishResponse{}, nil + + if !accepted { + if rejection != nil { + return nil, status.Error(codes.Internal, rejection.GetReason()) + } + return nil, status.Error(codes.Internal, fmt.Sprintf("all %d subscribers failed to accept the event", len(snapshot))) + } + return &chippb.PublishResponse{Results: []*chippb.PublishResult{{EventId: event.GetId()}}}, nil } func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch) (*chippb.PublishResponse, error) { + // Matches the write-service: an empty batch is a no-op success and never + // reaches the subscribers. + if len(batch.GetEvents()) == 0 { + return &chippb.PublishResponse{Results: []*chippb.PublishResult{}}, nil + } + snapshot := r.snapshotSubscribers() if len(snapshot) == 0 { // Acking events nothing carried would let a durable store delete them undelivered. - return nil, status.Error(codes.Unavailable, "no subscribers registered") + return nil, status.Error(codes.Internal, "no subscribers registered") } - // anyForwarded: at least one subscriber's RPC succeeded. detail: merged - // per-event results — an event is rejected only if every detail-providing - // subscriber rejected it; a subscriber accepting without per-event detail - // counts as accepting the whole batch. + // anyForwarded: at least one subscriber's RPC succeeded. anyBatchAccepted: + // a subscriber accepted the whole batch without per-event detail. detail: + // merged per-event results, index-aligned with the batch — an acceptance + // always wins over a rejection, and a rejection only fills an empty slot, + // so the outcome does not depend on subscriber completion order. var mu sync.Mutex anyForwarded := false - var detail []*chippb.PublishResult + anyBatchAccepted := false + detail := make([]*chippb.PublishResult, len(batch.GetEvents())) var group errgroup.Group group.SetLimit(forwardParallel) for _, sub := range snapshot { group.Go(func() error { - framework.L.Debug().Msgf("chip router forwarding batch to subscriber id=%s name=%s endpoint=%s", sub.id, sub.name, sub.endpoint) + logger.Debug().Msgf("chip router forwarding batch to subscriber id=%s name=%s endpoint=%s", sub.id, sub.name, sub.endpoint) forwardCtx, cancel := context.WithTimeout(ctx, forwardTimeout) defer cancel() resp, err := sub.client.PublishBatch(forwardCtx, batch) if err != nil { - framework.L.Error().Msgf("chip router failed to forward batch to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err) + logger.Error().Msgf("chip router failed to forward batch to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err) return nil } - framework.L.Debug().Msgf("chip router forwarded batch to subscriber id=%s", sub.id) + results := resp.GetResults() + if len(results) > 0 && len(results) != len(batch.GetEvents()) { + // The write-service returns one result per event in input order, so + // a short or long list cannot be aligned with the batch. Treat it + // as a failed forward rather than guess which events it covers. + logger.Warn().Msgf("chip router subscriber returned %d results for %d events, ignoring response id=%s name=%s endpoint=%s", len(results), len(batch.GetEvents()), sub.id, sub.name, sub.endpoint) + return nil + } + logger.Debug().Msgf("chip router forwarded batch to subscriber id=%s", sub.id) mu.Lock() defer mu.Unlock() anyForwarded = true - if results := resp.GetResults(); len(results) > 0 { - if detail == nil { - detail = results - return nil - } - // An accepting result overrides a rejection for the same event. - for i := range detail { - if i >= len(results) { - break - } - if detail[i].GetError() != nil && results[i].GetError() == nil { - detail[i] = results[i] - } + if len(results) == 0 { + // A detail-less success accepts the whole batch. + anyBatchAccepted = true + return nil + } + for i, res := range results { + if res.GetError() == nil { + detail[i] = res + } else if detail[i] == nil { + detail[i] = res } } return nil @@ -217,26 +284,24 @@ func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch // No subscriber accepted the batch: fail the RPC so every caller mode // retains and retries. if !anyForwarded { - return nil, status.Error(codes.Unavailable, fmt.Sprintf("all %d subscribers failed to accept the batch", len(snapshot))) + return nil, status.Error(codes.Internal, fmt.Sprintf("all %d subscribers failed to accept the batch", len(snapshot))) } + anyRejected := false results := make([]*chippb.PublishResult, 0, len(batch.GetEvents())) for i, ev := range batch.GetEvents() { result := &chippb.PublishResult{EventId: ev.GetId()} - if i < len(detail) && detail[i].GetError() != nil { + if !anyBatchAccepted && detail[i] != nil && detail[i].GetError() != nil { result.Error = detail[i].GetError() + anyRejected = true } results = append(results, result) } // Transactional callers resolve the whole batch from the RPC outcome // alone, so any unaccepted event must fail the RPC (all-or-nothing). - if batch.GetOptions().GetTransactionEnabled() { - for _, result := range results { - if result.Error != nil { - return nil, status.Error(codes.Unavailable, "not all events were accepted by a subscriber") - } - } + if batch.GetOptions().GetTransactionEnabled() && anyRejected { + return nil, status.Error(codes.Internal, "not all events were accepted by a subscriber") } return &chippb.PublishResponse{Results: results}, nil } @@ -270,13 +335,13 @@ func (r *router) handleSubscribers(w http.ResponseWriter, req *http.Request) { r.mu.Lock() if existing := r.findSubscriberLocked(body.Name, body.Endpoint); existing != nil { r.mu.Unlock() - framework.L.Info().Msgf("chip router reused subscriber id=%s name=%s endpoint=%s", existing.id, existing.name, existing.endpoint) + logger.Info().Msgf("chip router reused subscriber id=%s name=%s endpoint=%s", existing.id, existing.name, existing.endpoint) w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(registerSubscriberResponse{ID: existing.id}) return } - framework.L.Info().Msgf("chip router attempting to register subscriber name=%s endpoint=%s", body.Name, body.Endpoint) + logger.Info().Msgf("chip router attempting to register subscriber name=%s endpoint=%s", body.Name, body.Endpoint) conn, err := grpc.NewClient(body.Endpoint, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { r.mu.Unlock() @@ -292,7 +357,7 @@ func (r *router) handleSubscribers(w http.ResponseWriter, req *http.Request) { client: chippb.NewChipIngressClient(conn), } r.mu.Unlock() - framework.L.Info().Msgf("chip router registered subscriber id=%s name=%s endpoint=%s", id, body.Name, body.Endpoint) + logger.Info().Msgf("chip router registered subscriber id=%s name=%s endpoint=%s", id, body.Name, body.Endpoint) w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(registerSubscriberResponse{ID: id}) } @@ -324,7 +389,7 @@ func (r *router) handleSubscriberByID(w http.ResponseWriter, req *http.Request) r.mu.Unlock() if ok && sub.conn != nil { _ = sub.conn.Close() - framework.L.Info().Msgf("chip router unregistered subscriber id=%s name=%s endpoint=%s", id, sub.name, sub.endpoint) + logger.Info().Msgf("chip router unregistered subscriber id=%s name=%s endpoint=%s", id, sub.name, sub.endpoint) } w.WriteHeader(http.StatusNoContent) } diff --git a/framework/components/chiprouter/cmd/chip-router/main_test.go b/framework/components/chiprouter/cmd/chip-router/main_test.go index 3ce630356..10fd1e04b 100644 --- a/framework/components/chiprouter/cmd/chip-router/main_test.go +++ b/framework/components/chiprouter/cmd/chip-router/main_test.go @@ -2,7 +2,6 @@ package main import ( "context" - "sync" "testing" cepb "github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb" @@ -12,21 +11,26 @@ import ( "google.golang.org/grpc/status" ) +// stubPublishClient is immutable after construction, so it is safe for the +// router's concurrent forwards. A nil response makes Publish return the +// write-service default for a single event (one accepted result) and +// PublishBatch return a detail-less success (empty results). type stubPublishClient struct { - mu sync.Mutex - batches int err error response *chippb.PublishResponse } -func (s *stubPublishClient) Publish(context.Context, *cepb.CloudEvent, ...grpc.CallOption) (*chippb.PublishResponse, error) { - return &chippb.PublishResponse{}, nil +func (s *stubPublishClient) Publish(_ context.Context, in *cepb.CloudEvent, _ ...grpc.CallOption) (*chippb.PublishResponse, error) { + if s.err != nil { + return nil, s.err + } + if s.response != nil { + return s.response, nil + } + return &chippb.PublishResponse{Results: []*chippb.PublishResult{{EventId: in.GetId()}}}, nil } func (s *stubPublishClient) PublishBatch(context.Context, *chippb.CloudEventBatch, ...grpc.CallOption) (*chippb.PublishResponse, error) { - s.mu.Lock() - defer s.mu.Unlock() - s.batches++ if s.err != nil { return nil, s.err } @@ -36,22 +40,34 @@ func (s *stubPublishClient) PublishBatch(context.Context, *chippb.CloudEventBatc return &chippb.PublishResponse{}, nil } -func (s *stubPublishClient) batchCount() int { - s.mu.Lock() - defer s.mu.Unlock() - return s.batches -} - func ptr(b bool) *bool { return &b } +func eventOf(id string) *cepb.CloudEvent { + return &cepb.CloudEvent{Id: id, Source: "s", SpecVersion: "1.0", Type: "t"} +} + func batchOf(ids ...string) *chippb.CloudEventBatch { events := make([]*cepb.CloudEvent, 0, len(ids)) for _, id := range ids { - events = append(events, &cepb.CloudEvent{Id: id, Source: "s", SpecVersion: "1.0", Type: "t"}) + events = append(events, eventOf(id)) } return &chippb.CloudEventBatch{Events: events} } +func rejected(id, reason string) *chippb.PublishResult { + return &chippb.PublishResult{EventId: id, Error: &chippb.PublishError{Reason: reason}} +} + +func requireCode(t *testing.T, err error, want codes.Code) { + t.Helper() + if err == nil { + t.Fatalf("want %s error, got nil", want) + } + if status.Code(err) != want { + t.Errorf("want %s, got %v", want, status.Code(err)) + } +} + func TestPublishBatchPopulatesPerEventResults(t *testing.T) { r := &router{subscribers: map[string]*subscriber{ "sink": {id: "sink", client: &stubPublishClient{}}, @@ -74,31 +90,34 @@ func TestPublishBatchPopulatesPerEventResults(t *testing.T) { } } -func TestPublishBatchNoSubscribersIsUnavailable(t *testing.T) { +func TestPublishBatchEmptyBatchSucceedsWithoutSubscribers(t *testing.T) { + // The write-service treats an empty batch as a no-op success before any + // delivery attempt. r := &router{subscribers: map[string]*subscriber{}} - _, err := r.PublishBatch(t.Context(), batchOf("e1")) - if err == nil { - t.Fatal("want error with no subscribers, got nil") + resp, err := r.PublishBatch(t.Context(), batchOf()) + if err != nil { + t.Fatalf("PublishBatch: %v", err) } - if status.Code(err) != codes.Unavailable { - t.Errorf("want Unavailable, got %v", status.Code(err)) + if len(resp.Results) != 0 { + t.Errorf("want 0 results, got %d", len(resp.Results)) } } -func TestPublishBatchAllForwardsFailedIsUnavailable(t *testing.T) { +func TestPublishBatchNoSubscribersFailsWithInternal(t *testing.T) { + r := &router{subscribers: map[string]*subscriber{}} + + _, err := r.PublishBatch(t.Context(), batchOf("e1")) + requireCode(t, err, codes.Internal) +} + +func TestPublishBatchAllForwardsFailedFailsWithInternal(t *testing.T) { r := &router{subscribers: map[string]*subscriber{ "sink": {id: "sink", client: &stubPublishClient{err: context.DeadlineExceeded}}, }} - // Transactional callers resolve the whole batch from the RPC outcome alone. _, err := r.PublishBatch(t.Context(), batchOf("e1", "e2")) - if err == nil { - t.Fatal("want error when every forward fails, got nil") - } - if status.Code(err) != codes.Unavailable { - t.Errorf("want Unavailable, got %v", status.Code(err)) - } + requireCode(t, err, codes.Internal) } func TestPublishBatchAggregatesDownstreamRejections(t *testing.T) { @@ -108,10 +127,7 @@ func TestPublishBatchAggregatesDownstreamRejections(t *testing.T) { "sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{ Results: []*chippb.PublishResult{ {EventId: "e1"}, - {EventId: "e2", Error: &chippb.PublishError{ - ErrorCode: chippb.PublishErrorCode_PUBLISH_ERROR_CODE_VALIDATION_FAILED, - Reason: "invalid event", - }}, + rejected("e2", "invalid event"), }, }}}, }} @@ -131,6 +147,118 @@ func TestPublishBatchAggregatesDownstreamRejections(t *testing.T) { } } +func TestPublishBatchRejectionSurvivesAnotherSubscribersRPCFailure(t *testing.T) { + // A rejects e1 per-event; B's forward fails. e1 is still reported + // rejected in partial-delivery mode — it was accepted nowhere. + r := &router{subscribers: map[string]*subscriber{ + "a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{rejected("e1", "rejected by a")}, + }}}, + "b": {id: "b", client: &stubPublishClient{err: context.DeadlineExceeded}}, + }} + + resp, err := r.PublishBatch(t.Context(), batchOf("e1")) + if err != nil { + t.Fatalf("PublishBatch: %v", err) + } + if len(resp.Results) != 1 || resp.Results[0].Error == nil { + t.Errorf("want e1 rejected, got results=%+v", resp.Results) + } +} + +func TestPublishBatchAnAcceptingSubscriberOverridesARejection(t *testing.T) { + // Two detail-providing subscribers disagree: the event is rejected by one + // and accepted by the other — acceptance wins. + r := &router{subscribers: map[string]*subscriber{ + "a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{rejected("e1", "rejected by a")}, + }}}, + "b": {id: "b", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{{EventId: "e1"}}, + }}}, + }} + + resp, err := r.PublishBatch(t.Context(), batchOf("e1")) + if err != nil { + t.Fatalf("PublishBatch: %v", err) + } + if len(resp.Results) != 1 || resp.Results[0].Error != nil { + t.Errorf("want e1 accepted, got results=%+v", resp.Results) + } +} + +func TestPublishBatchDetaillessAcceptanceOverridesRejections(t *testing.T) { + // A rejects both events per-event; B accepts the whole batch without + // per-event detail. The batch was delivered to B, so both events are + // acknowledged regardless of completion order. + r := &router{subscribers: map[string]*subscriber{ + "a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{ + rejected("e1", "rejected by a"), + rejected("e2", "rejected by a"), + }, + }}}, + "b": {id: "b", client: &stubPublishClient{}}, + }} + + resp, err := r.PublishBatch(t.Context(), batchOf("e1", "e2")) + if err != nil { + t.Fatalf("PublishBatch: %v", err) + } + if len(resp.Results) != 2 { + t.Fatalf("want 2 results, got %d", len(resp.Results)) + } + for i, res := range resp.Results { + if res.Error != nil { + t.Errorf("results[%d].Error = %v, want nil (batch accepted by b)", i, res.Error) + } + } +} + +func TestPublishBatchResultsMismatchFromOnlySubscriberFailsWithInternal(t *testing.T) { + // The write-service returns one result per event, so a short list cannot + // be aligned with the batch: it counts as a failed forward, not as an + // acknowledgement of the events it does not cover. + r := &router{subscribers: map[string]*subscriber{ + "sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{{EventId: "e1"}}, + }}}, + }} + + _, err := r.PublishBatch(t.Context(), batchOf("e1", "e2")) + requireCode(t, err, codes.Internal) +} + +func TestPublishBatchResultsMismatchIsIgnoredInFavorOfFullDetail(t *testing.T) { + // A returns a short list (contract violation) and is dropped; B returns + // full detail rejecting e2, which must be the outcome. + r := &router{subscribers: map[string]*subscriber{ + "a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{{EventId: "e1"}}, + }}}, + "b": {id: "b", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{ + {EventId: "e1"}, + rejected("e2", "rejected by b"), + }, + }}}, + }} + + resp, err := r.PublishBatch(t.Context(), batchOf("e1", "e2")) + if err != nil { + t.Fatalf("PublishBatch: %v", err) + } + if len(resp.Results) != 2 { + t.Fatalf("want 2 results, got %d", len(resp.Results)) + } + if resp.Results[0].Error != nil { + t.Errorf("results[0].Error = %v, want nil", resp.Results[0].Error) + } + if resp.Results[1].Error == nil { + t.Error("results[1].Error = nil, want the rejection from b passed through") + } +} + func TestPublishBatchTransactionalPartialRejectionFailsRPC(t *testing.T) { // Transactional callers resolve the whole batch from the RPC outcome // alone, so a partial rejection must fail the RPC, not ack everything. @@ -138,7 +266,7 @@ func TestPublishBatchTransactionalPartialRejectionFailsRPC(t *testing.T) { "sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{ Results: []*chippb.PublishResult{ {EventId: "e1"}, - {EventId: "e2", Error: &chippb.PublishError{Reason: "rejected"}}, + rejected("e2", "rejected"), }, }}}, }} @@ -146,12 +274,7 @@ func TestPublishBatchTransactionalPartialRejectionFailsRPC(t *testing.T) { batch := batchOf("e1", "e2") batch.Options = &chippb.PublishOptions{TransactionEnabled: ptr(true)} _, err := r.PublishBatch(t.Context(), batch) - if err == nil { - t.Fatal("want error for a partial rejection under transaction_enabled, got nil") - } - if status.Code(err) != codes.Unavailable { - t.Errorf("want Unavailable, got %v", status.Code(err)) - } + requireCode(t, err, codes.Internal) } func TestPublishBatchTransactionalAllRejectedByDownstreamFailsRPC(t *testing.T) { @@ -160,8 +283,8 @@ func TestPublishBatchTransactionalAllRejectedByDownstreamFailsRPC(t *testing.T) r := &router{subscribers: map[string]*subscriber{ "sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{ Results: []*chippb.PublishResult{ - {EventId: "e1", Error: &chippb.PublishError{Reason: "rejected"}}, - {EventId: "e2", Error: &chippb.PublishError{Reason: "rejected"}}, + rejected("e1", "rejected"), + rejected("e2", "rejected"), }, }}}, }} @@ -169,12 +292,7 @@ func TestPublishBatchTransactionalAllRejectedByDownstreamFailsRPC(t *testing.T) batch := batchOf("e1", "e2") batch.Options = &chippb.PublishOptions{TransactionEnabled: ptr(true)} _, err := r.PublishBatch(t.Context(), batch) - if err == nil { - t.Fatal("want error when the subscriber rejected every event, got nil") - } - if status.Code(err) != codes.Unavailable { - t.Errorf("want Unavailable, got %v", status.Code(err)) - } + requireCode(t, err, codes.Internal) } func TestPublishBatchTransactionalAllAcceptedPasses(t *testing.T) { @@ -200,27 +318,110 @@ func TestPublishBatchTransactionalAllAcceptedPasses(t *testing.T) { } } -func TestPublishBatchAnAcceptingSubscriberOverridesADetaillessRejection(t *testing.T) { - // Two detail-providing subscribers disagree: the event is rejected by one - // and accepted by the other — acceptance wins. +func TestPublishReturnsOneResult(t *testing.T) { + // The write-service returns exactly one result for a single event. r := &router{subscribers: map[string]*subscriber{ - "a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{ - Results: []*chippb.PublishResult{ - {EventId: "e1", Error: &chippb.PublishError{Reason: "rejected by a"}}, - }, + "sink": {id: "sink", client: &stubPublishClient{}}, + }} + + resp, err := r.Publish(t.Context(), eventOf("e1")) + if err != nil { + t.Fatalf("Publish: %v", err) + } + if len(resp.Results) != 1 { + t.Fatalf("want 1 result, got %d", len(resp.Results)) + } + if resp.Results[0].EventId != "e1" { + t.Errorf("results[0].EventId = %q, want %q", resp.Results[0].EventId, "e1") + } + if resp.Results[0].Error != nil { + t.Errorf("results[0].Error = %v, want nil", resp.Results[0].Error) + } +} + +func TestPublishDetaillessSuccessAcceptsEvent(t *testing.T) { + // chainlink's chip-testsink subscriber returns an empty PublishResponse on + // success; the router must still acknowledge the event and echo its ID. + r := &router{subscribers: map[string]*subscriber{ + "sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{}}}, + }} + + resp, err := r.Publish(t.Context(), eventOf("e1")) + if err != nil { + t.Fatalf("Publish: %v", err) + } + if len(resp.Results) != 1 { + t.Fatalf("want 1 result, got %d", len(resp.Results)) + } + if resp.Results[0].EventId != "e1" { + t.Errorf("results[0].EventId = %q, want %q", resp.Results[0].EventId, "e1") + } + if resp.Results[0].Error != nil { + t.Errorf("results[0].Error = %v, want nil", resp.Results[0].Error) + } +} + +func TestPublishNoSubscribersFailsWithInternal(t *testing.T) { + // Acks without delivery would let a durable store delete the event undelivered. + r := &router{subscribers: map[string]*subscriber{}} + + _, err := r.Publish(t.Context(), eventOf("e1")) + requireCode(t, err, codes.Internal) +} + +func TestPublishAllForwardsFailedFailsWithInternal(t *testing.T) { + r := &router{subscribers: map[string]*subscriber{ + "sink": {id: "sink", client: &stubPublishClient{err: context.DeadlineExceeded}}, + }} + + _, err := r.Publish(t.Context(), eventOf("e1")) + requireCode(t, err, codes.Internal) +} + +func TestPublishAllSubscribersRejectedFailsWithInternal(t *testing.T) { + // Single-event Publish is all-or-nothing (the write-service treats it as + // transactional): a per-event rejection from every subscriber fails the RPC. + r := &router{subscribers: map[string]*subscriber{ + "sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{rejected("e1", "invalid event")}, }}}, - "b": {id: "b", client: &stubPublishClient{response: &chippb.PublishResponse{ - Results: []*chippb.PublishResult{ - {EventId: "e1"}, - }, + }} + + _, err := r.Publish(t.Context(), eventOf("e1")) + requireCode(t, err, codes.Internal) + if got := status.Convert(err).Message(); got != "invalid event" { + t.Errorf("error message = %q, want the subscriber's rejection reason %q", got, "invalid event") + } +} + +func TestPublishDetaillessAcceptanceOverridesARejection(t *testing.T) { + // A rejects the event per-event; B accepts with an empty response (no + // per-event detail) — the event was delivered to B, so Publish + // acknowledges it. + r := &router{subscribers: map[string]*subscriber{ + "a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{rejected("e1", "rejected by a")}, }}}, + "b": {id: "b", client: &stubPublishClient{response: &chippb.PublishResponse{}}}, }} - resp, err := r.PublishBatch(t.Context(), batchOf("e1")) + resp, err := r.Publish(t.Context(), eventOf("e1")) if err != nil { - t.Fatalf("PublishBatch: %v", err) + t.Fatalf("Publish: %v", err) } if len(resp.Results) != 1 || resp.Results[0].Error != nil { t.Errorf("want e1 accepted, got results=%+v", resp.Results) } } + +func TestPingMatchesWriteService(t *testing.T) { + r := &router{subscribers: map[string]*subscriber{}} + + resp, err := r.Ping(context.Background(), &chippb.EmptyRequest{}) + if err != nil { + t.Fatalf("Ping: %v", err) + } + if resp.GetMessage() != "pong" { + t.Errorf("Ping message = %q, want %q", resp.GetMessage(), "pong") + } +} diff --git a/framework/components/chiprouter/go.mod b/framework/components/chiprouter/go.mod index 18347708e..90c5dd5a6 100644 --- a/framework/components/chiprouter/go.mod +++ b/framework/components/chiprouter/go.mod @@ -9,6 +9,7 @@ require ( github.com/google/uuid v1.6.0 github.com/moby/moby/api v1.54.1 github.com/pkg/errors v0.9.1 + github.com/rs/zerolog v1.33.0 github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260915184316-2730f1867c92 github.com/smartcontractkit/chainlink-testing-framework/framework v0.15.8 github.com/testcontainers/testcontainers-go v0.42.0 @@ -60,7 +61,6 @@ require ( github.com/opencontainers/image-spec v1.1.1 // indirect github.com/pelletier/go-toml/v2 v2.2.4 // indirect github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect - github.com/rs/zerolog v1.33.0 // indirect github.com/shirou/gopsutil/v4 v4.26.3 // indirect github.com/sirupsen/logrus v1.9.4 // indirect github.com/stretchr/testify v1.12.1 // indirect From 9b58dfa863aea19357679d11d12802d0af432002 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Mon, 26 Oct 2026 11:54:44 -0400 Subject: [PATCH 7/9] chore(framework): add v0.16.10 changeset for chip-router write-service parity --- framework/.changeset/v0.16.10.md | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 framework/.changeset/v0.16.10.md diff --git a/framework/.changeset/v0.16.10.md b/framework/.changeset/v0.16.10.md new file mode 100644 index 000000000..0ee7a14ab --- /dev/null +++ b/framework/.changeset/v0.16.10.md @@ -0,0 +1,6 @@ +- `chip-router`: `PublishBatch` returns one `PublishResult` per event (previously an empty response, which made node durable emitters retransmit the whole stream), aggregating subscribers' per-event outcomes independent of completion order and honoring `transaction_enabled`; an empty batch is a no-op success +- `chip-router`: a subscriber response with a result count different from the batch size is ignored as a failed forward; an empty response still counts as a detail-less success +- `chip-router`: `Publish` (single event) is all-or-nothing like the chip-ingress write-service: it fails unless a subscriber accepted the event and returns exactly one `PublishResult` +- `chip-router`: failures return `codes.Internal` (the write-service's failure code), including when no subscribers are registered or none accepted, instead of acking undelivered events +- `chip-router`: implement `Ping` like the write-service +- `chip-router`: self-contained image build (builder `golang:1.26.6`, binary no longer imports the `framework` module); bump the `chainlink-common/pkg/chipingress` pin so `PublishResult.Error` is available From ffd92be702d2bf6a73a5856daa9033190148c3f2 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Mon, 26 Oct 2026 12:48:42 -0400 Subject: [PATCH 8/9] chore(chip-router): move release notes to the chiprouter module changeset chiprouter is its own Go module (tags framework/components/chiprouter/vX.Y.Z) and consumers pin it separately. release-go-module.yml reads .changeset/.md from the module directory and fails when it is missing, so the notes in framework/.changeset/v0.16.10.md could not release the router, while a framework release would have carried no framework change. Move them to chiprouter v1.0.5. --- .../v0.16.10.md => components/chiprouter/.changeset/v1.0.5.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename framework/{.changeset/v0.16.10.md => components/chiprouter/.changeset/v1.0.5.md} (100%) diff --git a/framework/.changeset/v0.16.10.md b/framework/components/chiprouter/.changeset/v1.0.5.md similarity index 100% rename from framework/.changeset/v0.16.10.md rename to framework/components/chiprouter/.changeset/v1.0.5.md From adaba38884e9ca53bea74ed55490ecb37c06cd8c Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Mon, 26 Oct 2026 12:48:56 -0400 Subject: [PATCH 9/9] fix(chip-router): always give a rejected Publish a non-empty error message A subscriber may reject an event without a PublishError.Reason, which made the router return a Publish error with an empty gRPC message. Prefix the message ('event rejected by subscribers: ') so it is never empty. --- .../chiprouter/cmd/chip-router/main.go | 2 +- .../chiprouter/cmd/chip-router/main_test.go | 20 +++++++++++++++++-- 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/framework/components/chiprouter/cmd/chip-router/main.go b/framework/components/chiprouter/cmd/chip-router/main.go index de6a7df73..70da3b490 100644 --- a/framework/components/chiprouter/cmd/chip-router/main.go +++ b/framework/components/chiprouter/cmd/chip-router/main.go @@ -210,7 +210,7 @@ func (r *router) Publish(ctx context.Context, event *cepb.CloudEvent) (*chippb.P if !accepted { if rejection != nil { - return nil, status.Error(codes.Internal, rejection.GetReason()) + return nil, status.Errorf(codes.Internal, "event rejected by subscribers: %s", rejection.GetReason()) } return nil, status.Error(codes.Internal, fmt.Sprintf("all %d subscribers failed to accept the event", len(snapshot))) } diff --git a/framework/components/chiprouter/cmd/chip-router/main_test.go b/framework/components/chiprouter/cmd/chip-router/main_test.go index 10fd1e04b..fffb226fc 100644 --- a/framework/components/chiprouter/cmd/chip-router/main_test.go +++ b/framework/components/chiprouter/cmd/chip-router/main_test.go @@ -389,8 +389,24 @@ func TestPublishAllSubscribersRejectedFailsWithInternal(t *testing.T) { _, err := r.Publish(t.Context(), eventOf("e1")) requireCode(t, err, codes.Internal) - if got := status.Convert(err).Message(); got != "invalid event" { - t.Errorf("error message = %q, want the subscriber's rejection reason %q", got, "invalid event") + if got, want := status.Convert(err).Message(), "event rejected by subscribers: invalid event"; got != want { + t.Errorf("error message = %q, want %q", got, want) + } +} + +func TestPublishRejectionWithoutReasonStillHasMessage(t *testing.T) { + // A subscriber may reject without a Reason; the caller must not see an + // empty gRPC error message. + r := &router{subscribers: map[string]*subscriber{ + "sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{rejected("e1", "")}, + }}}, + }} + + _, err := r.Publish(t.Context(), eventOf("e1")) + requireCode(t, err, codes.Internal) + if got := status.Convert(err).Message(); got == "" { + t.Error("error message is empty, want a non-empty message") } }