Fix Go timeout scan and receive errors

Author: ktsaou

.agents/sow/done/SOW-0016-20260610-client-call-timeout-and-abort.md

@@ -4,7 +4,7 @@
 
 Status: completed
 
-Sub-state: implementation and validation completed on 2026-06-11; `SOW-0015` remains paused and can be resumed after this SOW is committed.
+Sub-state: regression repair completed on 2026-06-11 after CI failures on commit `aaec3d5`; `SOW-0015` remains paused and can be resumed after this repair is committed.
 
 ## Requirements
 
@@ -375,4 +375,53 @@ None.
 
 ## Regression Log
 
-None yet.
+### 2026-06-11
+
+CI failures after commit `aaec3d5` reopened this SOW:
+
+- Static Analysis run `27336192003`, job `Go Static Analysis (src/go)`, failed because `gosec` exited with status 1 after uploading SARIF. Local reproduction with `gosec -fmt=json -exclude=G404 ./...` under `src/go` reports `G115` at `src/go/pkg/netipc/service/raw/client_unix.go:198`: integer overflow conversion `int64 -> uint32` in the SHM receive polling timeout calculation.
+- CodeQL run `27336192249`, job `Analyze Go Windows`, failed in the MSYS2 build step while running Go tests. The failing test was `TestSessionReceiveRejectsMalformedChunks/continuation_recv_disconnect` in `src/go/pkg/netipc/transport/windows/pipe_edge_test.go`; CI observed `peer disconnected` while the existing test expected the old wrapped continuation receive error.
+
+Why previous validation missed it:
+
+- The local Go validation did not run `gosec`; the static-analysis workflow did.
+- The final `win11` validation did run Windows Go tests, but the CI failure is a Windows named-pipe close timing variation. The changed framing code returned raw continuation receive errors, so one platform/timing path exposed a direct `ErrDisconnected` instead of the pre-existing wrapped `ErrRecv` contract.
+
+Repair plan:
+
+- Keep timeout and abort errors distinct through chunk continuation receives.
+- Restore pre-SOW behavior for ordinary continuation receive failures by wrapping them as `ErrRecv("continuation recv: ...")`.
+- Replace duplicated timeout polling conversions with bounded helpers that prove the narrowed value is below the poll cap before converting to `uint32`.
+
+Validation plan:
+
+- Re-run `gosec -fmt=json -exclude=G404 ./...` under `src/go`.
+- Re-run `cd src/go && go test ./pkg/netipc/...` locally.
+- Re-run the failing Windows package/test on `win11` with `MSYSTEM=MSYS`.
+- Re-run `git diff --check`.
+
+Repair results:
+
+- Updated `src/go/pkg/netipc/transport/internal/framing/receive.go` so chunk continuation receive errors are wrapped as `ErrRecv("continuation recv: ...")` unless the transport explicitly marks them as timeout/abort control errors.
+- Updated POSIX UDS and Windows Named Pipe receive config to propagate only timeout and abort receive errors raw.
+- Replaced duplicated Go SHM timeout narrowing logic in `src/go/pkg/netipc/service/raw/client_unix.go` and `src/go/pkg/netipc/service/raw/client_windows.go` with `boundedClientWaitMs()`, which caps the value before the justified `G115`-suppressed conversion.
+- Updated Windows named-pipe receive polling to use the same bounded conversion pattern.
+
+Validation results:
+
+- `cd src/go && go test ./pkg/netipc/...`: passed.
+- `cd src/go && go vet ./...`: passed.
+- `cd src/go && staticcheck ./...`: passed.
+- Exact CI-style `gosec -quiet -fmt sarif -out gosec.sarif -exclude=G404 ./...` passed for `src/go`, `tests/fixtures/go`, and `bench/drivers/go`.
+- Verbose local `gosec -fmt=json -exclude=G404 ./...` under `src/go` now reports zero issues; it still reports the pre-existing assembly-stub SSA warning for `pkg/netipc/transport/posix/shm_pause_amd64.go`, but the CI-style SARIF command exits successfully with zero findings.
+- `win11` MSYS CodeQL-style loop passed: `CGO_ENABLED=0 go test ./...` in `src/go`, `tests/fixtures/go`, and `bench/drivers/go`.
+- `win11` focused failure reproduction passed 50 times: `CGO_ENABLED=0 go test ./pkg/netipc/transport/windows -run 'TestSessionReceiveRejectsMalformedChunks/continuation_recv_disconnect' -count=50`.
+- `git diff --check`: passed.
+
+Regression artifact maintenance:
+
+- `AGENTS.md`: no update needed; workflow and project guardrails did not change.
+- Runtime project skills: no update needed; this was a narrow code regression repair, not a reusable workflow change.
+- Specs and end-user/operator docs: no update needed; public timeout/abort behavior remains as documented.
+- End-user/operator skills: no update needed; integrator guidance remains correct.
+- SOW lifecycle: SOW was reopened from `done/`, repaired, validated, marked completed, and moved back to `done/` with the repair commit.

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

@@ -194,12 +194,7 @@ func (c *Client) transportReceiveWithControl(timeoutMs uint32, abortCh <-chan st
 				if remaining <= 0 {
 					return protocol.Header{}, nil, protocol.ErrTimeout
 				}
-				if remaining < time.Duration(waitMs)*time.Millisecond {
-					waitMs = uint32((remaining + time.Millisecond - 1) / time.Millisecond)
-					if waitMs == 0 {
-						waitMs = 1
-					}
-				}
+				waitMs = boundedClientWaitMs(remaining, waitMs)
 			}
 
 			mlen, err := c.shm.ShmReceive(scratch, waitMs)

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

@@ -182,12 +182,7 @@ func (c *Client) transportReceiveWithControl(timeoutMs uint32, abortCh <-chan st
 				if remaining <= 0 {
 					return protocol.Header{}, nil, protocol.ErrTimeout
 				}
-				if remaining < time.Duration(waitMs)*time.Millisecond {
-					waitMs = uint32((remaining + time.Millisecond - 1) / time.Millisecond)
-					if waitMs == 0 {
-						waitMs = 1
-					}
-				}
+				waitMs = boundedClientWaitMs(remaining, waitMs)
 			}
 
 			mlen, err := c.shm.WinShmReceive(scratch, waitMs)

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

@@ -8,6 +8,8 @@
 // Pure Go — no cgo. Works with CGO_ENABLED=0.
 package raw
 
+import "time"
+
 // Poll/receive timeout for server loops (ms). Controls shutdown detection latency.
 const serverPollTimeoutMs = 100
 
@@ -16,6 +18,20 @@ const ClientCallTimeoutDefaultMs uint32 = 30000
 
 const clientAbortPollMs uint32 = 100
 
+func boundedClientWaitMs(remaining time.Duration, pollCapMs uint32) uint32 {
+	if remaining <= 0 {
+		return 0
+	}
+	if remaining >= time.Duration(pollCapMs)*time.Millisecond {
+		return pollCapMs
+	}
+	waitMs := uint32((remaining + time.Millisecond - 1) / time.Millisecond) // #nosec G115 -- remaining is positive and below pollCapMs here.
+	if waitMs == 0 {
+		return 1
+	}
+	return waitMs
+}
+
 // ---------------------------------------------------------------------------
 //  Client state (shared across platforms)
 // ---------------------------------------------------------------------------

src/go/pkg/netipc/transport/internal/framing/receive.go

@@ -23,6 +23,7 @@ type Receiver struct {
 	Recv                func([]byte) (int, error)
 	EnsurePacketScratch func(*[]byte, int) []byte
 	OnRecvError         func(error)
+	PropagateRecvError  func(error) bool
 
 	ErrLimitExceeded func(string) error
 	ErrProtocol      func(string) error
@@ -47,6 +48,7 @@ type SessionReceiveConfig struct {
 	Recv                func([]byte) (int, error)
 	EnsurePacketScratch func(*[]byte, int) []byte
 	IsRecvDisconnect    func(error) bool
+	PropagateRecvError  func(error) bool
 	FailAllInflight     func()
 
 	ErrLimitExceeded func(string) error
@@ -74,6 +76,7 @@ func SessionReceive(config SessionReceiveConfig, buf []byte) (protocol.Header, [
 		PacketBuf:           config.PacketBuf,
 		Recv:                config.Recv,
 		EnsurePacketScratch: config.EnsurePacketScratch,
+		PropagateRecvError:  config.PropagateRecvError,
 		OnRecvError: func(err error) {
 			if config.IsRecvDisconnect(err) {
 				config.FailAllInflight()
@@ -256,7 +259,10 @@ func (r Receiver) receiveOneChunk(
 	cn, err := r.Recv(pktBuf)
 	if err != nil {
 		r.noteRecvError(err)
-		return err
+		if r.PropagateRecvError != nil && r.PropagateRecvError(err) {
+			return err
+		}
+		return r.ErrRecv("continuation recv: " + err.Error())
 	}
 	if cn < protocol.HeaderSize {
 		return r.ErrChunk("continuation too short")

src/go/pkg/netipc/transport/posix/uds_receive.go

@@ -151,11 +151,14 @@ func (s *Session) ReceiveTimeout(buf []byte, timeoutMs uint32, abortCh <-chan st
 		Recv:                    recv,
 		EnsurePacketScratch:     ensureScratchBuf,
 		IsRecvDisconnect:        func(err error) bool { return errors.Is(err, ErrRecv) },
-		FailAllInflight:         s.failAllInflight,
-		ErrLimitExceeded:        func(msg string) error { return wrapErr(ErrLimitExceeded, msg) },
-		ErrProtocol:             func(msg string) error { return wrapErr(ErrProtocol, msg) },
-		ErrChunk:                func(msg string) error { return wrapErr(ErrChunk, msg) },
-		ErrUnknownMsgID:         func(msg string) error { return wrapErr(ErrUnknownMsgID, msg) },
-		ErrRecv:                 func(msg string) error { return wrapErr(ErrRecv, msg) },
+		PropagateRecvError: func(err error) bool {
+			return errors.Is(err, ErrTimeout) || errors.Is(err, ErrAborted)
+		},
+		FailAllInflight:  s.failAllInflight,
+		ErrLimitExceeded: func(msg string) error { return wrapErr(ErrLimitExceeded, msg) },
+		ErrProtocol:      func(msg string) error { return wrapErr(ErrProtocol, msg) },
+		ErrChunk:         func(msg string) error { return wrapErr(ErrChunk, msg) },
+		ErrUnknownMsgID:  func(msg string) error { return wrapErr(ErrUnknownMsgID, msg) },
+		ErrRecv:          func(msg string) error { return wrapErr(ErrRecv, msg) },
 	}, buf)
 }

src/go/pkg/netipc/transport/windows/pipe_receive.go

@@ -11,6 +11,8 @@ import (
 	"github.com/netdata/plugin-ipc/go/pkg/netipc/transport/internal/framing"
 )
 
+const receiveAbortPollMs uint32 = 100
+
 // Receive reads one logical message. buf is a scratch buffer.
 func (s *Session) Receive(buf []byte) (protocol.Header, []byte, error) {
 	return s.ReceiveTimeout(buf, 0, nil)
@@ -40,12 +42,15 @@ func (s *Session) ReceiveTimeout(buf []byte, timeoutMs uint32, abortCh <-chan st
 		Recv:                    recv,
 		EnsurePacketScratch:     ensurePipeScratchBuf,
 		IsRecvDisconnect:        func(err error) bool { return errors.Is(err, ErrDisconnected) },
-		FailAllInflight:         s.failAllInflight,
-		ErrLimitExceeded:        func(msg string) error { return wrapErr(ErrLimitExceeded, msg) },
-		ErrProtocol:             func(msg string) error { return wrapErr(ErrProtocol, msg) },
-		ErrChunk:                func(msg string) error { return wrapErr(ErrChunk, msg) },
-		ErrUnknownMsgID:         func(msg string) error { return wrapErr(ErrUnknownMsgID, msg) },
-		ErrRecv:                 func(msg string) error { return wrapErr(ErrRecv, msg) },
+		PropagateRecvError: func(err error) bool {
+			return errors.Is(err, ErrTimeout) || errors.Is(err, ErrAborted)
+		},
+		FailAllInflight:  s.failAllInflight,
+		ErrLimitExceeded: func(msg string) error { return wrapErr(ErrLimitExceeded, msg) },
+		ErrProtocol:      func(msg string) error { return wrapErr(ErrProtocol, msg) },
+		ErrChunk:         func(msg string) error { return wrapErr(ErrChunk, msg) },
+		ErrUnknownMsgID:  func(msg string) error { return wrapErr(ErrUnknownMsgID, msg) },
+		ErrRecv:          func(msg string) error { return wrapErr(ErrRecv, msg) },
 	}, buf)
 }
 
@@ -75,7 +80,7 @@ func (w receiveWait) waitMs() (uint32, error) {
 		}
 	}
 
-	waitMs := uint32(100)
+	waitMs := receiveAbortPollMs
 	if w.infinite {
 		return waitMs, nil
 	}
@@ -84,13 +89,21 @@ func (w receiveWait) waitMs() (uint32, error) {
 	if remaining <= 0 {
 		return 0, ErrTimeout
 	}
-	if remaining < time.Duration(waitMs)*time.Millisecond {
-		waitMs = uint32((remaining + time.Millisecond - 1) / time.Millisecond)
-		if waitMs == 0 {
-			waitMs = 1
-		}
+	return boundedReceiveWaitMs(remaining, waitMs), nil
+}
+
+func boundedReceiveWaitMs(remaining time.Duration, pollCapMs uint32) uint32 {
+	if remaining <= 0 {
+		return 0
+	}
+	if remaining >= time.Duration(pollCapMs)*time.Millisecond {
+		return pollCapMs
+	}
+	waitMs := uint32((remaining + time.Millisecond - 1) / time.Millisecond) // #nosec G115 -- remaining is positive and below pollCapMs here.
+	if waitMs == 0 {
+		return 1
 	}
-	return waitMs, nil
+	return waitMs
 }
 
 func (s *Session) rawRecvWithTimeout(buf []byte, wait receiveWait) (int, error) {