diff --git a/framework/components/chiprouter/.changeset/v1.0.5.md b/framework/components/chiprouter/.changeset/v1.0.5.md new file mode 100644 index 000000000..0ee7a14ab --- /dev/null +++ b/framework/components/chiprouter/.changeset/v1.0.5.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 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 516d5b71f..70da3b490 100644 --- a/framework/components/chiprouter/cmd/chip-router/main.go +++ b/framework/components/chiprouter/cmd/chip-router/main.go @@ -16,14 +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" - - "github.com/smartcontractkit/chainlink-testing-framework/framework" + "google.golang.org/grpc/status" ) +// 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" @@ -65,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) } } @@ -108,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(): @@ -131,59 +152,158 @@ 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.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))) + } + 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 { - return &chippb.PublishResponse{}, nil + // Acking events nothing carried would let a durable store delete them undelivered. + return nil, status.Error(codes.Internal, "no subscribers registered") } + // 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 + 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() - _, 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) + 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 + } + 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 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 + } } - framework.L.Debug().Msgf("chip router forwarded batch to subscriber id=%s", sub.id) return nil }) } _ = group.Wait() - return &chippb.PublishResponse{}, nil + + // No subscriber accepted the batch: fail the RPC so every caller mode + // retains and retries. + if !anyForwarded { + 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 !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() && anyRejected { + return nil, status.Error(codes.Internal, "not all events were accepted by a subscriber") + } + return &chippb.PublishResponse{Results: results}, nil } func (r *router) handleHealth(w http.ResponseWriter, req *http.Request) { @@ -215,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() @@ -237,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}) } @@ -269,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 new file mode 100644 index 000000000..fffb226fc --- /dev/null +++ b/framework/components/chiprouter/cmd/chip-router/main_test.go @@ -0,0 +1,443 @@ +package main + +import ( + "context" + "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" +) + +// 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 { + err error + response *chippb.PublishResponse +} + +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) { + if s.err != nil { + return nil, s.err + } + if s.response != nil { + return s.response, nil + } + return &chippb.PublishResponse{}, nil +} + +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, 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{}}, + }} + + 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 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{}} + + resp, err := r.PublishBatch(t.Context(), batchOf()) + if err != nil { + t.Fatalf("PublishBatch: %v", err) + } + if len(resp.Results) != 0 { + t.Errorf("want 0 results, got %d", len(resp.Results)) + } +} + +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}}, + }} + + _, err := r.PublishBatch(t.Context(), batchOf("e1", "e2")) + requireCode(t, err, codes.Internal) +} + +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"}, + rejected("e2", "invalid event"), + }, + }}}, + }} + + 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 (accepted)", resp.Results[0].Error) + } + if resp.Results[1].Error == nil { + t.Error("results[1].Error = nil, want the downstream rejection passed through") + } +} + +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. + r := &router{subscribers: map[string]*subscriber{ + "sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{ + Results: []*chippb.PublishResult{ + {EventId: "e1"}, + rejected("e2", "rejected"), + }, + }}}, + }} + + batch := batchOf("e1", "e2") + batch.Options = &chippb.PublishOptions{TransactionEnabled: ptr(true)} + _, err := r.PublishBatch(t.Context(), batch) + requireCode(t, err, codes.Internal) +} + +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{ + rejected("e1", "rejected"), + rejected("e2", "rejected"), + }, + }}}, + }} + + batch := batchOf("e1", "e2") + batch.Options = &chippb.PublishOptions{TransactionEnabled: ptr(true)} + _, err := r.PublishBatch(t.Context(), batch) + requireCode(t, err, codes.Internal) +} + +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 TestPublishReturnsOneResult(t *testing.T) { + // The write-service returns exactly one result for a single event. + r := &router{subscribers: map[string]*subscriber{ + "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")}, + }}}, + }} + + _, err := r.Publish(t.Context(), eventOf("e1")) + requireCode(t, err, codes.Internal) + 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") + } +} + +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.Publish(t.Context(), eventOf("e1")) + if err != nil { + 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 c31193136..90c5dd5a6 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,12 @@ 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/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 golang.org/x/sync v0.22.0 - google.golang.org/grpc v1.83.1 + google.golang.org/grpc v1.83.2 ) require ( @@ -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 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= 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 => ../../