fix(chip-router): match atlas write-service Publish/PublishBatch semantics

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

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)
 }