Reduce NetIPC platform duplication

Author: ktsaou

.agents/sow/current/SOW-0015-20260605-codacy-scope-and-maintainability.md

@@ -669,6 +669,41 @@ SonarCloud duplication composition for Netdata PR #22649:
 
 The duplication is a real POSIX/Windows paired-implementation signal, not unresolved Sonar line findings. It will be handled after this review-finding fix is committed and re-vendored.
 
+### 2026-06-07 - SonarCloud Duplication Reduction
+
+Live Netdata PR #22649 recheck before this increment:
+
+- GitHub review threads: eight total, eight resolved, zero open.
+- SonarCloud issue API: zero unresolved issues on the PR.
+- SonarCloud hotspot API: zero unresolved hotspots on the PR.
+- SonarCloud quality gate status: failed only because new-code duplication is 10.5%, above the configured 3% threshold.
+- Latest SonarCloud duplication component tree still reports 1,826 duplicated new lines across 29 files.
+
+Selected low-risk production-source refactors:
+
+- Extract Go POSIX/Windows transport receive framing from `uds_receive.go` and `pipe_receive.go` into a shared internal framing receiver. This targets the top two SonarCloud contributors, 189 duplicated lines each.
+- Extract the Go raw-service per-session request/dispatch/response loop from `server_unix.go` and `server_windows.go` into shared raw-service helpers while leaving OS-specific accept, readiness, receive, send, and SHM cleanup code local. This targets the next two contributors, 160 and 159 duplicated lines.
+- Extract Go POSIX/Windows transport send chunking and HELLO/HELLO_ACK protocol helpers into shared internal framing helpers while keeping socket/pipe I/O and session construction platform-local. This targets the next Go transport contributors: handshake and send pairs.
+- Extract C client refresh, raw-call envelope validation, and retry/reconnect policy into common service helpers with platform callbacks. This targets the POSIX/Windows C client-call and client lifecycle duplication without changing transport send/receive functions.
+
+Risk controls:
+
+- Do not merge POSIX and Windows accept loops; they differ in readiness, listener shutdown, and SHM setup behavior.
+- Keep transport-specific receive/send as callbacks so OS-specific error and timeout semantics remain local.
+- Validate POSIX Go packages locally and Windows Go packages on Win11 after the refactor.
+- Validate C common helper changes on POSIX and Win11 because both platform service client paths now call common retry/raw-call helpers.
+
+Validation completed for this duplication-reduction increment:
+
+- `git diff --check`: passed.
+- `cd src/go && go test -count=1 ./pkg/netipc/transport/internal/framing ./pkg/netipc/transport/posix ./pkg/netipc/service/raw`: passed.
+- `cd src/go && go test -count=1 ./pkg/netipc/...`: passed.
+- `cmake --build build`: passed.
+- `/usr/bin/ctest --test-dir build --output-on-failure`: 46/46 tests passed.
+- Win11 temp-copy Go validation: `cd src/go && go test -count=1 ./pkg/netipc/transport/windows ./pkg/netipc/service/raw`: passed.
+- Win11 temp-copy C validation: MSYS CMake build of `test_win_service`, `test_win_service_extra`, and `test_win_service_payload_limits` passed; CTest for those three tests passed.
+- `bash .agents/sow/audit.sh`: passed.
+
 ## Validation
 
 Acceptance criteria evidence:

src/go/pkg/netipc/service/raw/server.go

@@ -1,6 +1,11 @@
 package raw
 
-import "sync/atomic"
+import (
+	"errors"
+	"sync/atomic"
+
+	"github.com/netdata/plugin-ipc/go/pkg/netipc/protocol"
+)
 
 func (s *Server) dispatchSingle(methodCode uint16, request []byte, responseBuf []byte) (int, error) {
 	if methodCode != s.expectedMethodCode || s.handler == nil {
@@ -26,3 +31,186 @@ func serverNotePayloadCapacity(target *atomic.Uint32, payloadLen uint32) {
 		}
 	}
 }
+
+type serverReceiveAction uint8
+
+const (
+	serverReceiveOK serverReceiveAction = iota
+	serverReceiveContinue
+	serverReceiveStop
+)
+
+type serverSessionOps struct {
+	maxRequestPayloadBytes  uint32
+	maxResponsePayloadBytes uint32
+	receive                 func([]byte) (protocol.Header, []byte, serverReceiveAction)
+	send                    func(*protocol.Header, []byte, *[]byte) error
+	close                   func()
+}
+
+func (s *Server) handleServerSession(ops serverSessionOps) {
+	recvBuf := make([]byte, protocol.HeaderSize+int(ops.maxRequestPayloadBytes))
+	respBuf := make([]byte, int(ops.maxResponsePayloadBytes))
+	itemRespBuf := make([]byte, int(ops.maxResponsePayloadBytes))
+	msgBuf := make([]byte, int(ops.maxResponsePayloadBytes)+protocol.HeaderSize)
+
+	defer ops.close()
+
+	for s.running.Load() {
+		hdr, payload, action := ops.receive(recvBuf)
+		switch action {
+		case serverReceiveContinue:
+			continue
+		case serverReceiveStop:
+			return
+		}
+
+		if hdr.Kind != protocol.KindRequest {
+			return
+		}
+
+		respHdr, responseLen, closeAfterSend := s.handleServerRequest(
+			hdr, payload, respBuf, itemRespBuf, ops.maxResponsePayloadBytes)
+		if err := ops.send(&respHdr, respBuf[:responseLen], &msgBuf); err != nil {
+			return
+		}
+		if closeAfterSend {
+			return
+		}
+	}
+}
+
+func (s *Server) handleServerRequest(
+	hdr protocol.Header,
+	payload []byte,
+	respBuf []byte,
+	itemRespBuf []byte,
+	maxResponsePayloadBytes uint32,
+) (protocol.Header, int, bool) {
+	if payloadLen, err := checkedLookupU32(len(payload)); err == nil {
+		serverNotePayloadCapacity(&s.learnedRequestPayloadBytes, payloadLen)
+	}
+
+	if !s.methodSupported(hdr.Code) {
+		return serverUnsupportedResponseHeader(hdr), 0, false
+	}
+
+	responseLen, isBatch, dispatchErr := s.dispatchServerResponse(hdr, payload, respBuf, itemRespBuf)
+	respHdr := serverResponseHeader(hdr)
+
+	if dispatchErr == nil {
+		if responseLen32, err := checkedLookupU32(responseLen); err == nil {
+			serverNotePayloadCapacity(&s.learnedResponsePayloadBytes, responseLen32)
+		}
+		respHdr.TransportStatus = protocol.StatusOK
+		if isBatch {
+			respHdr.Flags = protocol.FlagBatch
+			respHdr.ItemCount = hdr.ItemCount
+		} else {
+			respHdr.ItemCount = 1
+		}
+		return respHdr, responseLen, false
+	}
+
+	respHdr.ItemCount = 1
+	responseLen = 0
+	switch {
+	case errors.Is(dispatchErr, protocol.ErrOverflow):
+		if maxResponsePayloadBytes >= ^uint32(0)/2 {
+			serverNotePayloadCapacity(&s.learnedResponsePayloadBytes, ^uint32(0))
+		} else {
+			serverNotePayloadCapacity(&s.learnedResponsePayloadBytes, maxResponsePayloadBytes*2)
+		}
+		respHdr.TransportStatus = protocol.StatusLimitExceeded
+		return respHdr, responseLen, true
+	case errors.Is(dispatchErr, errHandlerFailed):
+		respHdr.TransportStatus = protocol.StatusInternalError
+	default:
+		respHdr.TransportStatus = protocol.StatusBadEnvelope
+	}
+
+	return respHdr, responseLen, false
+}
+
+func (s *Server) dispatchServerResponse(
+	hdr protocol.Header,
+	payload []byte,
+	respBuf []byte,
+	itemRespBuf []byte,
+) (int, bool, error) {
+	isBatch := (hdr.Flags&protocol.FlagBatch != 0) && hdr.ItemCount >= 1
+	if !isBatch {
+		responseLen, err := s.dispatchSingle(hdr.Code, payload, respBuf)
+		if err != nil {
+			return 0, false, err
+		}
+		if responseLen < 0 || responseLen > len(respBuf) {
+			return 0, false, protocol.ErrOverflow
+		}
+		return responseLen, false, nil
+	}
+
+	var bb protocol.BatchBuilder
+	bb.Reset(respBuf, hdr.ItemCount)
+	for i := uint32(0); i < hdr.ItemCount; i++ {
+		itemData, err := protocol.BatchItemGet(payload, hdr.ItemCount, i)
+		if err != nil {
+			return 0, true, err
+		}
+
+		itemResultLen, err := s.dispatchSingle(hdr.Code, itemData, itemRespBuf)
+		if err != nil {
+			return 0, true, err
+		}
+		if itemResultLen < 0 || itemResultLen > len(itemRespBuf) {
+			return 0, true, protocol.ErrOverflow
+		}
+		if err := bb.Add(itemRespBuf[:itemResultLen]); err != nil {
+			return 0, true, err
+		}
+	}
+
+	responseLen, _ := bb.Finish()
+	return responseLen, true, nil
+}
+
+func serverResponseHeader(hdr protocol.Header) protocol.Header {
+	return protocol.Header{
+		Kind:      protocol.KindResponse,
+		Code:      hdr.Code,
+		MessageID: hdr.MessageID,
+	}
+}
+
+func serverUnsupportedResponseHeader(hdr protocol.Header) protocol.Header {
+	respHdr := serverResponseHeader(hdr)
+	respHdr.TransportStatus = protocol.StatusUnsupported
+	respHdr.ItemCount = 1
+	return respHdr
+}
+
+func serverEncodeSharedResponse(
+	respHdr *protocol.Header,
+	payload []byte,
+	msgBuf *[]byte,
+) ([]byte, error) {
+	payloadLen, err := checkedLookupU32(len(payload))
+	if err != nil {
+		return nil, err
+	}
+	msgLen := protocol.HeaderSize + len(payload)
+	if len(*msgBuf) < msgLen {
+		*msgBuf = make([]byte, msgLen)
+	}
+	msg := (*msgBuf)[:msgLen]
+
+	respHdr.Magic = protocol.MagicMsg
+	respHdr.Version = protocol.Version
+	respHdr.HeaderLen = protocol.HeaderLen
+	respHdr.PayloadLen = payloadLen
+	respHdr.Encode(msg[:protocol.HeaderSize])
+	if len(payload) > 0 {
+		copy(msg[protocol.HeaderSize:], payload)
+	}
+	return msg, nil
+}