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