Conversation
The RPC framework could only run over connections it established itself: QUIC/WebTransport, or a TCP listener it owned. Two needs break that assumption. A clusters API exposed over a WebSocket carrying somebody else's envelope protocol hands rpc a connection it must not dial, listen on, or close. And TCP clients want the leanest possible path to the cluster API, without an HTTP layer in the way. Both reduce to running over a single, externally supplied, message oriented pipe. MessageConn is that seam: Send, Recv, Close over discrete byte messages, with msgmux layering stream multiplexing on top to recover the callback and streaming semantics the framework needs. Transport: - Callstreams share the one session wrapping the connection; callbacks route by OID through a session-scoped registry. handleCallStream takes a callbackRouter so WebTransport keeps its per-call session and its existing wire format, which deployed clusters depend on. - Every stream on a message session opens with an opRequest, including callbacks (opInlineCall), so one accept loop routes the whole session and a peer that dialed outbound can still be called back into. - Capabilities minted on a connection nobody can dial carry no address; following one reuses the session it arrived on. - ServeMessageConn and ClientFromMessageConn are the public entry points, with WithMaxFrameSize to fit an envelope's own size limit. - The WebSocket transport is a wsConn MessageConn over coder/websocket; tcp:// is the same session over a length-prefixed TLS socket with no HTTP at all. rpcSession/rpcStream is shared by WebTransport and msgmux. Auth: - Authenticator.Authenticate took an *http.Request, so cloudauth, oidcauth and outboard could not run on a message transport. It now takes a transport-neutral Credentials struct; CredentialsFromRequest fills it for the HTTP paths. opRequest.Bearer carries a token where there is no header, and the authenticator's identity supersedes the capability-signature identity when a token is present. Protocol: - PROTOCOL.md documents the wire format end to end so a second implementation needs no breaking change. opRequest.Version gates the stream; absent or zero reads as version 1, and a mismatch is rejected with a reply naming the supported range. Also fixes a pre-existing race on heldCapability.lastContact, which concurrent calls against one capability all wrote unsynchronized, and an authorization denial surfacing as "unknown response status" on the message transport.
📝 WalkthroughWalkthroughThis PR changes the RPC authentication contract from Comment |
![miren-code-agent[bot]](https://avatars.githubusercontent.com/u/174767976?s=60&v=4){kind=small}
There was a problem hiding this comment.
🍪 biscuit:
This is a draft, so I'm judging whether it's ready for human review. It introduces a full MessageConn transport layer for RPC — multiplexer (msgmux), three backends (mem, TCP, WebSocket), and all the client/server plumbing — plus a well-structured CBOR protocol on top of ed25519-signed frames. The architecture is clean, the new Authenticator interface refactor (*http.Request → *Credentials) is a meaningful improvement, and the test suite is genuinely thorough: unary, streaming, bidirectional callbacks, concurrent callstreams, panic propagation, authorization denial, bearer tokens, version gating, and frame-size splitting are all covered.
A few things I want to flag before this moves to merge review:
ws:// connects over TLS despite its name. In message_ws.go, dialWSMessageConn is called for both ws:// and wss:// remotes — and it always prepends "wss://" when dialing. So ws:// does not mean unencrypted; it's just an alias for wss:// at this layer. That's fine behaviour, but the scheme naming is misleading. The messageSchemes slice in transport_msg.go and the docs should clarify this, or ws:// should be dropped and only wss:// accepted, so operators don't assume they're getting a plaintext path.
accept channel capacity of 16 is a hidden backpressure point. In msgmux.go the accept channel is created with a buffer of 16. If the sessionRouter.run loop is slow dispatching streams (or a slow handler holds up the goroutine schedule), a peer sending more than 16 concurrent SYN frames will have subsequent SYNs silently dropped in handleFrame (the case s.accept <- st: select will fall through to case <-s.done:). The frame is then gone — not queued elsewhere, not retried — so the server will never serve those streams. This is a correctness risk under heavy concurrency. Consider either blocking in handleFrame (like a real mux does) or using a larger / dynamically growing queue, and documenting the limit clearly.
Dead session is not cleared after reconnect. In controlSession, when a dialed session's underlying connection dies, ctrlSess is never set to nil. The next call to controlSession returns the closed session, which will fail immediately, rather than redialing. Once a session is dead — whether its readPump exited or its conn closed — there is no path back to the dial() branch. For adopted sessions this is intentional and fine, but for mem://, ws://, and tcp:// clients the documented "lost capability → reresolve" recovery path assumes the transport can recover, which it currently cannot. The simplest fix would be to nil out ctrlSess in the shutdown path and have controlSession detect a closed session and re-dial.
freshTimestamp is one-sided. It rejects requests older than 10 minutes but doesn't reject timestamps in the future. A clock-skewed or replayed-ahead frame can be accepted unconditionally. This is the same issue that exists on the HTTP path, so it isn't a regression — but now that replay protection is being relied on for a new transport it's worth noting for a follow-up.
Everything else — signing, frame splitting, FIN/RST semantics, the inline-capability registry, the perCallRouter/inlineRegistry split — looks solid and well-tested. The draft is ready to graduate to human review with the caveats above noted.
🍪 full review note · comment /biscuit review to run biscuit again.
| streams: make(map[uint32]*msgStream), | ||
| nextID: start, | ||
| idStep: 2, | ||
| accept: make(chan *msgStream, 16), |
There was a problem hiding this comment.
The accept channel has a fixed capacity of 16. In handleFrame, if all 16 slots are full the select races s.done, not a block — so the 17th inbound SYN is silently dropped and the stream is lost. Under a burst of concurrent client-opened streams this can silently lose accepted streams. Consider blocking the send (dropping the select entirely and using a plain s.accept <- st) if you're willing to stall the read pump briefly, or growing the buffer significantly and documenting the limit.
🤖 Prompt for AI Agents
In pkg/rpc/msgmux.go, the accept channel is
created with a buffer of 16 (line 78: `accept:
make(chan *msgStream, 16)`). In handleFrame around
line 177-183, when a new stream arrives and the
accept channel is full, the select falls through
to `case <-s.done:` and the stream is silently
dropped — the SYN frame is consumed but the
caller's AcceptStream never sees it. Fix this by
blocking on the send instead of selecting, e.g.
replace the select block with a plain `s.accept <-
st` (while still checking s.done to avoid a
goroutine leak), or document clearly what the
capacity bound means for callers and increase it
to something that won't be hit in realistic usage.
| t.mu.Lock() | ||
| defer t.mu.Unlock() | ||
|
|
||
| if t.ctrlSess != nil { |
There was a problem hiding this comment.
Once a dialed session's connection closes (its readPump exits and calls shutdown), ctrlSess is never cleared. The next call to controlSession returns the already-closed session, which will fail immediately on OpenStreamSync, rather than falling through to the dial branch and reconnecting. For mem://, ws://, and tcp:// clients this means a dropped connection is unrecoverable. Consider setting ctrlSess = nil when the session shuts down so subsequent calls can re-dial.
🤖 Prompt for AI Agents
In pkg/rpc/transport_msg.go, controlSession
(around line 38) checks `t.ctrlSess != nil` and
returns the cached session immediately. When that
session's underlying connection dies its readPump
calls shutdown, but ctrlSess is never set back to
nil. So the next call to controlSession returns a
closed session that fails immediately, rather than
re-dialing. Fix this by hooking into session
closure: after `startAcceptLoop` launches, watch
for the session's `done` channel in a goroutine,
and when it fires, lock t.mu, compare and clear
t.ctrlSess, then unlock — so the next
controlSession call can re-dial using t.dial.
| cfg := tlsCfg.Clone() | ||
| cfg.NextProtos = []string{"http/1.1"} | ||
|
|
||
| c, _, err := websocket.Dial(ctx, "wss://"+remote+wsMessagePath, &websocket.DialOptions{ |
There was a problem hiding this comment.
dialWSMessageConn always dials wss:// regardless of whether the caller passed ws:// or wss:// as the scheme. So ws:// is effectively an alias for TLS-over-WebSocket, not an unencrypted connection. This is fine security-wise, but the scheme name is misleading — operators might expect ws:// to mean plaintext. Either remove ws:// from the accepted scheme list (keeping only wss://) or add a clear note in the doc comment that ws:// is treated identically to wss://.
| if err != nil { | ||
| return fmt.Errorf("invalid timestamp: %w", err) | ||
| } | ||
| if time.Since(t) > authFreshness { |
There was a problem hiding this comment.
freshTimestamp rejects requests older than authFreshness (10 min) but accepts any future timestamp, including one far in the future. A client could pre-sign requests with a timestamp 9 minutes in the future, gaining a ~19-minute window in which they remain valid and survive a clock rollback replay. Worth at least bounding: abs(time.Since(t)) > authFreshness.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 19
🧹 Nitpick comments (8)
pkg/rpc/server.go (1)
1020-1032: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winConsider
errors.Asso wrapped errors keep their category/code.Direct type assertions only match the outermost error.
cond.Wrapreturns cond errors unchanged so the common path works, but any handler returningfmt.Errorf("...: %w", cond.ErrNotFound{...})falls back to a bare message with emptycategory/codeon the wire.♻️ Suggested change
func errorFields(err error) (msg, category, code string) { - if emsg, ok := err.(ErrorMessage); ok { + var emsg ErrorMessage + if errors.As(err, &emsg) { msg = emsg.ErrorMessage() } else { msg = err.Error() } - if ecat, ok := err.(ErrorCategory); ok { + var ecat ErrorCategory + if errors.As(err, &ecat) { category = ecat.ErrorCategory() } - if ecode, ok := err.(ErrorCode); ok { + var ecode ErrorCode + if errors.As(err, &ecode) { code = ecode.ErrorCode() } return }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/rpc/server.go` around lines 1020 - 1032, Update errorFields to use errors.As when extracting ErrorMessage, ErrorCategory, and ErrorCode so wrapped errors are unwrapped and retain their metadata. Preserve the existing fallback to err.Error() and empty fields when no matching interfaces are found.pkg/rpc/message_net_test.go (1)
68-79: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueMake
startedsignalling idempotent.
close(m.started)panics if the handler is ever invoked twice on the same instance (client retry, or a future test reusing the value). Async.Oncekeeps the signal semantics without the sharp edge.🛡️ Proposed hardening
type blockingMU struct { started chan struct{} + once sync.Once } func (m *blockingMU) RegisterUpdates(ctx context.Context, call *example.MeterUpdatesRegisterUpdates) error { - close(m.started) + m.once.Do(func() { close(m.started) })🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/rpc/message_net_test.go` around lines 68 - 79, Make the started notification in blockingMU.RegisterUpdates idempotent by adding a sync.Once field to blockingMU and guarding close(m.started) with it. Preserve the existing channel signal and handler wait behavior for every invocation.pkg/rpc/message_adopt_test.go (2)
31-34: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueOptional: collapse the counter into
*atomic.Int64.The
envelopes *int64+countMu *sync.Mutexpair can be a single shared*atomic.Int64, dropping a field and the lock/unlock inSend.♻️ Proposed simplification
// envelopes counts payloads that made a round trip through the wrapper, so // a test can assert traffic actually crossed the boundary. - envelopes *int64 - countMu *sync.Mutex + envelopes *atomic.Int64 }- c.countMu.Lock() - *c.envelopes++ - c.countMu.Unlock() + c.envelopes.Add(1)func (c *envelopeConn) count() int64 { - c.countMu.Lock() - defer c.countMu.Unlock() - return *c.envelopes + return c.envelopes.Load() }Also applies to: 53-59, 92-96
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/rpc/message_adopt_test.go` around lines 31 - 34, In the test wrapper’s envelope counter, replace the shared envelopes *int64 and countMu *sync.Mutex fields with a single shared *atomic.Int64. Update Send and all initialization or assertions around the wrapper to use atomic load/add operations, removing the mutex locking and unlocking while preserving the round-trip count behavior.
256-263: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winFixed sleeps before asserting async completion across the new transport tests. Every one of these sites sleeps a hardcoded interval and then asserts the final state, which is simultaneously slow on fast machines and flaky on loaded CI; the shared root cause is asserting on a timer instead of polling until the expectation holds (
require.EventuallyWithT).
pkg/rpc/message_adopt_test.go#L256-L263: replace the 500ms sleep with a polled assertion onvalstaken undermu.pkg/rpc/message_net_test.go#L128-L132: replace the 100ms "wait for the goroutine running Close" sleep with a polled assertion onup.closedunderup.mu.pkg/rpc/message_net_test.go#L155-L162: replace the one-second sleep with a polled assertion onvalsundermu.pkg/rpc/transport_mem_test.go#L83-L87: replace the 100ms sleep with a polled assertion onup.closedunderup.mu.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/rpc/message_adopt_test.go` around lines 256 - 263, Replace timer-based async assertions with require.EventuallyWithT polling: in pkg/rpc/message_adopt_test.go:256-263 poll vals while holding mu; in pkg/rpc/message_net_test.go:128-132 poll up.closed while holding up.mu; in pkg/rpc/message_net_test.go:155-162 poll vals while holding mu; and in pkg/rpc/transport_mem_test.go:83-87 poll up.closed while holding up.mu. Remove the corresponding sleeps while preserving each existing expected value/state assertion.pkg/rpc/state_ws.go (1)
45-49: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUse
errors.Isfor theErrServerClosedcomparison.ServeTLSmay return a wrapped error, in which case the!=check logs a spurious error on normal shutdown.♻️ Proposed change
- if serr := srv.ServeTLS(tcpLn, "", ""); serr != nil && serr != http.ErrServerClosed { + if serr := srv.ServeTLS(tcpLn, "", ""); serr != nil && !errors.Is(serr, http.ErrServerClosed) { s.log.Error("websocket/tcp listener stopped", "error", serr) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/rpc/state_ws.go` around lines 45 - 49, Update the ServeTLS error check in the goroutine to use errors.Is when comparing serr with http.ErrServerClosed, while preserving the existing logging behavior for other errors.pkg/rpc/message_tcp.go (1)
48-59: 🗄️ Data Integrity & Integration | 🔵 Trivial | 💤 Low valueGuard the outbound frame length before narrowing to
uint32.uint32(len(b))silently truncates for payloads over 4 GiB, which would write a bogus header and desynchronize the peer's framing rather than failing loudly. msgmux chunking makes this practically unreachable today, but the check is one line and keeps the invariant local to the framer.🛡️ Proposed guard
func (t *tcpConn) Send(b []byte) error { + if uint64(len(b)) > math.MaxUint32 { + return fmt.Errorf("rpc: frame of %d bytes exceeds the framing limit", len(b)) + } + t.wmu.Lock() defer t.wmu.Unlock()Requires adding
"math"to the imports.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/rpc/message_tcp.go` around lines 48 - 59, Update tcpConn.Send to reject payloads whose length exceeds the uint32 frame-length limit before calling binary.BigEndian.PutUint32. Return an error without writing any header or payload, while preserving the existing locking and write behavior for valid payloads; add the required math import.Source: Linters/SAST tools
pkg/rpc/message_mem.go (1)
95-111: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueClient end is not closed if the router exits.
dialhands the callerclientEndand spawns the router onserverEnd; since both ends share the samedonechannel, this is fine for teardown, but a duplicateregisterunder the same name silently replaces the previous server with no signal to the caller. Consider returning an error (or logging) on name collision so a mis-wired test/host doesn't silently steal traffic.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/rpc/message_mem.go` around lines 95 - 111, Update memRegistry registration logic to detect an existing server name before replacing it, and return an error (or log through the established mechanism) on duplicate registration. Preserve the existing registered server and its active client connections when a collision occurs; use the register method and server-name registry symbols to implement the check.pkg/rpc/message.go (1)
13-25: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDocument that
Close()must unblock a pendingRecv().
msgSession.Close()only callsconn.Close()and depends on that unblocking thereadPump's in-flightRecv()to trigger session shutdown. The interface doc only describesRecv's behavior on peer closure, not on a localClose()call — worth spelling out explicitly for third-party implementers (per PROTOCOL.md's "Implementing another peer" section), otherwise a conforming-lookingMessageConncould leak the read-pump goroutine onClose().📝 Proposed doc clarification
// MessageConn is the pure message interface a message-oriented backend // implements: a reliable, ordered, bidirectional, point-to-point pipe of // discrete byte messages between two peers. The framework layers a stream // multiplexer (msgmux) on top to recover the rpcSession semantics that // callbacks and streaming require. // // Recv blocks until a message is available and returns io.EOF once the peer // has closed the connection and no buffered messages remain. +// +// Close must also cause any blocked Send or Recv on this connection to return +// promptly with an error — msgSession.Close relies on this to unblock its +// read loop and complete session shutdown. type MessageConn interface {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/rpc/message.go` around lines 13 - 25, Update the MessageConn interface documentation to explicitly require that Close unblocks any pending Recv call, causing it to return promptly with an appropriate error. Keep the existing Recv behavior for peer closure and buffered messages unchanged, and make the requirement clear for third-party implementations.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pkg/rpc/client.go`:
- Around line 774-781: Handle and propagate errors from both CBOR encodes in the
control-stream request path around the encoder setup: check the prelude encoding
result when prelude is non-nil and the args encoding result, returning
immediately on failure instead of proceeding to dec.Decode. Preserve the
existing conditional prelude behavior and ensure the caller receives the
original encode or stream error.
- Around line 104-115: Update the doc comment for NetworkClient.setupTransport
to reference the actual transport implementation files, replacing the
nonexistent transport_ws.go reference with transport_msg.go and message_ws.go.
Simplify the switch by removing the redundant explicit transportH3 case and
retaining the shared H3 fallback in default.
In `@pkg/rpc/inline_router.go`:
- Around line 133-145: In the stream-processing switch for “call” requests,
consume and discard the call arguments before replying and continuing when
lookup(rs.OID) fails or mm.Handler is nil. Reuse the existing args decoding
type/path from the normal call flow, and handle decode failure consistently;
alternatively close the stream on these error paths to prevent returning a
desynchronized connection to the pool.
In `@pkg/rpc/message_adopt_test.go`:
- Around line 396-413: Strengthen the “respects a frame size below the payload
size” subtest by asserting that conn.count() exceeds the envelope count expected
for an unsplit round trip, rather than only checking it is nonzero. Keep the
existing temperature assertions and use the connection’s count to verify
WithMaxFrameSize(64) actually causes payload splitting.
In `@pkg/rpc/message_tcp.go`:
- Around line 136-158: Update NetworkClient.setupTCPTransport so its
msgOpTransport dial callback accepts and propagates the caller’s context to a
context-aware TLS dial, or otherwise applies a bounded dial timeout. Preserve
the existing TLS configuration and newTCPConn setup while ensuring an
unresponsive peer cannot block indefinitely.
In `@pkg/rpc/message_ws.go`:
- Around line 75-93: The WebSocket transport must bound reads using the session
frame cap plus envelope headroom instead of disabling the limit. In
pkg/rpc/message_ws.go lines 75-93, thread the session’s maxFrame into the server
accept path and apply the computed cap in serveWSUpgrade; in
pkg/rpc/message_ws.go lines 61-63, apply the same cap on the client dial path
and update the comment to describe the bound. Reuse the calculation established
by newTCPConn rather than hard-coding a value.
In `@pkg/rpc/msgmux.go`:
- Around line 293-323: Update CancelRead and the peer-RST failure path to mark
the stream locally closed and immediately call removeStream, matching the
existing receive-RST teardown behavior. Ensure Write uses that local-closure
state to return errSessionClosed after cancellation or reset instead of sending
frames. Preserve normal writes and existing stream-locking semantics.
- Around line 101-120: Update OpenStreamSync so the SYN frame send honors ctx
cancellation and deadlines instead of blocking indefinitely in sendFrame.
Preserve the existing closed-session, stream registration, and cleanup behavior,
and ensure cancellation returns the context error while removing the abandoned
stream; account for a late send completion without letting it return the stream
after cancellation.
In `@pkg/rpc/server_session.go`:
- Around line 382-395: Update the bearer handling in the session authentication
flow around authenticator.Authenticate so a present but invalid or expired
req.Bearer returns an authentication error instead of falling back to the signed
identity. Preserve the existing signed-identity path when no bearer token is
supplied, and retain successful bearer identity handling.
- Around line 274-283: Update msgRef to release s.mu before calling writeReply:
perform the object lookup and reference increment while locked, capture the
success or error response, then unlock and write the reply outside the critical
section. Preserve the existing responses for known and unknown capabilities
while avoiding defer-based locking across network I/O.
- Around line 220-240: Update the resolver restoration branch around
res.ReconstructFromState so a missing resolver (ok == false) returns an
appropriate error before assignCapability is called, rather than passing a nil
interface. Set category from rs.Category in this branch as required, and apply
the same missing-resolver guard and category assignment in the reresolve
handler.
In `@pkg/rpc/server.go`:
- Line 505: Fix the inverted rpc-contact-addr fallback in all three handlers at
pkg/rpc/server.go lines 505, 603, and 700: update each guard so
s.state.contactAddr() is used only when ca is empty, preserving any
client-supplied contact address.
In `@pkg/rpc/signing.go`:
- Around line 56-67: Update freshTimestamp to reject timestamps that are older
than authFreshness or ahead of the current time by the permitted freshness
window, so future-dated requests cannot bypass the replay bound. Preserve the
existing RFC3339Nano parsing and error handling, and keep valid timestamps
within the bounded window accepted.
In `@pkg/rpc/state_ws.go`:
- Around line 30-33: Update the http.Server initialization in stateWebSocket
handling to set ReadHeaderTimeout to an appropriate time.Duration, adding the
time import as needed. Leave ReadTimeout and WriteTimeout unset so long-lived
hijacked WebSocket connections remain unaffected.
In `@pkg/rpc/transport_mem_test.go`:
- Around line 99-109: Add mutex protection to both stream accumulator tests in
pkg/rpc/transport_mem_test.go: at lines 99-109, guard vals appends in
stream.StreamRecv and the r.Equal read with a sync.Mutex; at lines 139-154, give
each goroutine’s vals its own mutex, guard appends and reads, and store a copied
slice in got[i] rather than aliasing the live accumulator.
In `@pkg/rpc/transport_msg.go`:
- Around line 319-365: The msgUnaryCall retry loop currently re-resolves
capabilities and retries indefinitely when the server repeatedly returns
unknown-capability. Limit this recovery path to one re-resolve and retry
attempt, then return cond.NotFound if unknown-capability persists or
re-resolution fails, while preserving normal handling for all other reply
statuses.
- Around line 83-117: Update msgOpTransport.roundTrip to honor ctx cancellation
during request encoding and reply decoding, not only in OpenStreamSync. Add the
same CancelRead-on-ctx.Done guard used by the callstream path, ensure the
watcher is cleaned up with the stream, and return the resulting decode error
when cancellation interrupts a blocked peer response.
- Around line 38-56: Update the session-termination handling around
controlSession and its accept-loop goroutine to clear t.ctrlSess when a dialed
control session ends, while guarding the reset so adopted sessions with dial ==
nil remain pinned and continue returning the existing “not redialable” error.
Ensure the cleared session can be redialed on the next controlSession call.
In `@pkg/rpc/transport_wt.go`:
- Around line 66-69: Update isRetryableTransportError to return true only for
the explicitly intended retryable QUIC application-error codes, rather than
every *quic.ApplicationError; preserve non-application-error handling and ensure
Call and handleCallStreamDialError stop retrying for other codes.
---
Nitpick comments:
In `@pkg/rpc/message_adopt_test.go`:
- Around line 31-34: In the test wrapper’s envelope counter, replace the shared
envelopes *int64 and countMu *sync.Mutex fields with a single shared
*atomic.Int64. Update Send and all initialization or assertions around the
wrapper to use atomic load/add operations, removing the mutex locking and
unlocking while preserving the round-trip count behavior.
- Around line 256-263: Replace timer-based async assertions with
require.EventuallyWithT polling: in pkg/rpc/message_adopt_test.go:256-263 poll
vals while holding mu; in pkg/rpc/message_net_test.go:128-132 poll up.closed
while holding up.mu; in pkg/rpc/message_net_test.go:155-162 poll vals while
holding mu; and in pkg/rpc/transport_mem_test.go:83-87 poll up.closed while
holding up.mu. Remove the corresponding sleeps while preserving each existing
expected value/state assertion.
In `@pkg/rpc/message_mem.go`:
- Around line 95-111: Update memRegistry registration logic to detect an
existing server name before replacing it, and return an error (or log through
the established mechanism) on duplicate registration. Preserve the existing
registered server and its active client connections when a collision occurs; use
the register method and server-name registry symbols to implement the check.
In `@pkg/rpc/message_net_test.go`:
- Around line 68-79: Make the started notification in blockingMU.RegisterUpdates
idempotent by adding a sync.Once field to blockingMU and guarding
close(m.started) with it. Preserve the existing channel signal and handler wait
behavior for every invocation.
In `@pkg/rpc/message_tcp.go`:
- Around line 48-59: Update tcpConn.Send to reject payloads whose length exceeds
the uint32 frame-length limit before calling binary.BigEndian.PutUint32. Return
an error without writing any header or payload, while preserving the existing
locking and write behavior for valid payloads; add the required math import.
In `@pkg/rpc/message.go`:
- Around line 13-25: Update the MessageConn interface documentation to
explicitly require that Close unblocks any pending Recv call, causing it to
return promptly with an appropriate error. Keep the existing Recv behavior for
peer closure and buffered messages unchanged, and make the requirement clear for
third-party implementations.
In `@pkg/rpc/server.go`:
- Around line 1020-1032: Update errorFields to use errors.As when extracting
ErrorMessage, ErrorCategory, and ErrorCode so wrapped errors are unwrapped and
retain their metadata. Preserve the existing fallback to err.Error() and empty
fields when no matching interfaces are found.
In `@pkg/rpc/state_ws.go`:
- Around line 45-49: Update the ServeTLS error check in the goroutine to use
errors.Is when comparing serr with http.ErrServerClosed, while preserving the
existing logging behavior for other errors.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 4c820637-2a43-4cec-8e24-ffc370cda3d3
📒 Files selected for processing (35)
pkg/cloudauth/rpc_authenticator.gopkg/cloudauth/rpc_authenticator_cert_test.gopkg/oidcauth/authenticator.gopkg/oidcauth/authenticator_test.gopkg/oidcauth/composite.gopkg/oidcauth/composite_test.gopkg/outboard/outboard_test.gopkg/outboard/token_auth.gopkg/rpc/PROTOCOL.mdpkg/rpc/actor.gopkg/rpc/authenticator.gopkg/rpc/authenticator_test.gopkg/rpc/call.gopkg/rpc/client.gopkg/rpc/inline.gopkg/rpc/inline_router.gopkg/rpc/message.gopkg/rpc/message_adopt_test.gopkg/rpc/message_mem.gopkg/rpc/message_net_test.gopkg/rpc/message_tcp.gopkg/rpc/message_version_test.gopkg/rpc/message_ws.gopkg/rpc/msgmux.gopkg/rpc/msgmux_test.gopkg/rpc/server.gopkg/rpc/server_session.gopkg/rpc/signing.gopkg/rpc/state.gopkg/rpc/state_ws.gopkg/rpc/stream/stream_test.gopkg/rpc/transport.gopkg/rpc/transport_mem_test.gopkg/rpc/transport_msg.gopkg/rpc/transport_wt.go
| // setupTransport configures the client's round-tripper and callstream dialer | ||
| // based on the selected transport. The h3 and ws implementations live in | ||
| // transport_wt.go and transport_ws.go respectively. | ||
| func (c *NetworkClient) setupTransport() { | ||
| c.htr.Logger = c.State.log.With("module", "rpc-call") | ||
| c.htr.TLSClientConfig = c.tlsCfg | ||
| c.htr.QUICConfig = &DefaultQUICConfig | ||
| c.htr.Dial = func(ctx context.Context, addr string, tlsCfg *tls.Config, cfg *quic.Config) (*quic.Conn, error) { | ||
| uaddr, err := resolveUDPAddr(ctx, "udp", addr) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| setTLSConfigServerName(tlsCfg, uaddr, addr) | ||
|
|
||
| return c.transport.DialEarly(ctx, uaddr, tlsCfg, cfg) | ||
| switch c.transportKind { | ||
| case transportMsg: | ||
| c.setupMsgTransport() | ||
| case transportH3: | ||
| c.setupH3Transport() | ||
| default: | ||
| c.setupH3Transport() | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Stale doc comment: there is no transport_ws.go. The message-transport setup lives in transport_msg.go (with the WS specifics in message_ws.go). Also the explicit transportH3 case and default are identical.
✏️ Proposed change
-// setupTransport configures the client's round-tripper and callstream dialer
-// based on the selected transport. The h3 and ws implementations live in
-// transport_wt.go and transport_ws.go respectively.
+// setupTransport configures the client's round-tripper and callstream dialer
+// based on the selected transport. The implementations live in transport_wt.go
+// (HTTP/3 + WebTransport) and transport_msg.go (message transports).
func (c *NetworkClient) setupTransport() {
- switch c.transportKind {
- case transportMsg:
+ if c.transportKind == transportMsg {
c.setupMsgTransport()
- case transportH3:
- c.setupH3Transport()
- default:
- c.setupH3Transport()
+ return
}
+
+ c.setupH3Transport()
}As per coding guidelines: "Only add comments when they provide valuable context or explain why something is done ... ensure function/method comments explain purpose and important side effects".
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // setupTransport configures the client's round-tripper and callstream dialer | |
| // based on the selected transport. The h3 and ws implementations live in | |
| // transport_wt.go and transport_ws.go respectively. | |
| func (c *NetworkClient) setupTransport() { | |
| c.htr.Logger = c.State.log.With("module", "rpc-call") | |
| c.htr.TLSClientConfig = c.tlsCfg | |
| c.htr.QUICConfig = &DefaultQUICConfig | |
| c.htr.Dial = func(ctx context.Context, addr string, tlsCfg *tls.Config, cfg *quic.Config) (*quic.Conn, error) { | |
| uaddr, err := resolveUDPAddr(ctx, "udp", addr) | |
| if err != nil { | |
| return nil, err | |
| } | |
| setTLSConfigServerName(tlsCfg, uaddr, addr) | |
| return c.transport.DialEarly(ctx, uaddr, tlsCfg, cfg) | |
| switch c.transportKind { | |
| case transportMsg: | |
| c.setupMsgTransport() | |
| case transportH3: | |
| c.setupH3Transport() | |
| default: | |
| c.setupH3Transport() | |
| } | |
| // setupTransport configures the client's round-tripper and callstream dialer | |
| // based on the selected transport. The implementations live in transport_wt.go | |
| // (HTTP/3 + WebTransport) and transport_msg.go (message transports). | |
| func (c *NetworkClient) setupTransport() { | |
| if c.transportKind == transportMsg { | |
| c.setupMsgTransport() | |
| return | |
| } | |
| c.setupH3Transport() | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/rpc/client.go` around lines 104 - 115, Update the doc comment for
NetworkClient.setupTransport to reference the actual transport implementation
files, replacing the nonexistent transport_ws.go reference with transport_msg.go
and message_ws.go. Simplify the switch by removing the redundant explicit
transportH3 case and retaining the shared H3 fallback in default.
Source: Coding guidelines
| enc := cbor.NewEncoder(ctrl) | ||
| // On the message transport the control stream is also the operation stream, | ||
| // so it must lead with the opRequest the server's router decodes; the HTTP | ||
| // transports pass a nil prelude. | ||
| if prelude != nil { | ||
| enc.Encode(prelude) | ||
| } | ||
| enc.Encode(args) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Encode errors on the control stream are dropped. If the prelude or args fail to encode (or the stream is already broken), the code proceeds to dec.Decode and the caller sees an opaque read failure instead of the actual cause. On the message transport the prelude is the operation frame, so a dropped error here means the server never gets a dispatchable request.
🐛 Proposed fix
if prelude != nil {
- enc.Encode(prelude)
+ if err := enc.Encode(prelude); err != nil {
+ return err
+ }
}
- enc.Encode(args)
+ if err := enc.Encode(args); err != nil {
+ return err
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| enc := cbor.NewEncoder(ctrl) | |
| // On the message transport the control stream is also the operation stream, | |
| // so it must lead with the opRequest the server's router decodes; the HTTP | |
| // transports pass a nil prelude. | |
| if prelude != nil { | |
| enc.Encode(prelude) | |
| } | |
| enc.Encode(args) | |
| enc := cbor.NewEncoder(ctrl) | |
| // On the message transport the control stream is also the operation stream, | |
| // so it must lead with the opRequest the server's router decodes; the HTTP | |
| // transports pass a nil prelude. | |
| if prelude != nil { | |
| if err := enc.Encode(prelude); err != nil { | |
| return err | |
| } | |
| } | |
| if err := enc.Encode(args); err != nil { | |
| return err | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/rpc/client.go` around lines 774 - 781, Handle and propagate errors from
both CBOR encodes in the control-stream request path around the encoder setup:
check the prelude encoding result when prelude is non-nil and the args encoding
result, returning immediately on failure instead of proceeding to dec.Decode.
Preserve the existing conditional prelude behavior and ensure the caller
receives the original encode or stream error.
| switch rs.Kind { | ||
| case "call": | ||
| c, iface, ok := lookup(rs.OID) | ||
| if !ok { | ||
| _ = enc.Encode(refResponse{Status: "error", Error: "unknown capability"}) | ||
| continue | ||
| } | ||
|
|
||
| mm := iface.methods[rs.Method] | ||
| if mm.Handler == nil { | ||
| _ = enc.Encode(refResponse{Status: "error", Error: "unknown method"}) | ||
| continue | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
Error paths leave the call's args unread, desynchronizing the pooled stream.
The peer writes streamRequest and then the args (inlineClient.Call, pkg/rpc/inline.go Lines 173-187). On the unknown-capability/unknown-method paths this loop replies and continues without consuming the args value, so the next dec.Decode(&rs) reads the args payload as a streamRequest. Worse, the client treats refResponse{Status:"error"} as a normal remote error and returns the stream to its pool, so the corrupted stream is reused for later calls.
Decode-and-discard the args before continuing (or close the stream on these paths).
🐛 Proposed fix
case "call":
c, iface, ok := lookup(rs.OID)
if !ok {
+ var discard cbor.RawMessage
+ if err := dec.Decode(&discard); err != nil {
+ return
+ }
_ = enc.Encode(refResponse{Status: "error", Error: "unknown capability"})
continue
}
mm := iface.methods[rs.Method]
if mm.Handler == nil {
+ var discard cbor.RawMessage
+ if err := dec.Decode(&discard); err != nil {
+ return
+ }
_ = enc.Encode(refResponse{Status: "error", Error: "unknown method"})
continue
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| switch rs.Kind { | |
| case "call": | |
| c, iface, ok := lookup(rs.OID) | |
| if !ok { | |
| _ = enc.Encode(refResponse{Status: "error", Error: "unknown capability"}) | |
| continue | |
| } | |
| mm := iface.methods[rs.Method] | |
| if mm.Handler == nil { | |
| _ = enc.Encode(refResponse{Status: "error", Error: "unknown method"}) | |
| continue | |
| } | |
| switch rs.Kind { | |
| case "call": | |
| c, iface, ok := lookup(rs.OID) | |
| if !ok { | |
| var discard cbor.RawMessage | |
| if err := dec.Decode(&discard); err != nil { | |
| return | |
| } | |
| _ = enc.Encode(refResponse{Status: "error", Error: "unknown capability"}) | |
| continue | |
| } | |
| mm := iface.methods[rs.Method] | |
| if mm.Handler == nil { | |
| var discard cbor.RawMessage | |
| if err := dec.Decode(&discard); err != nil { | |
| return | |
| } | |
| _ = enc.Encode(refResponse{Status: "error", Error: "unknown method"}) | |
| continue | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/rpc/inline_router.go` around lines 133 - 145, In the stream-processing
switch for “call” requests, consume and discard the call arguments before
replying and continuing when lookup(rs.OID) fails or mm.Handler is nil. Reuse
the existing args decoding type/path from the normal call flow, and handle
decode failure consistently; alternatively close the stream on these error paths
to prevent returning a desynchronized connection to the pool.
| // A frame cap below the payload size forces msgmux to split a write across | ||
| // several messages, which is what an envelope with its own size limit needs. | ||
| t.Run("respects a frame size below the payload size", func(t *testing.T) { | ||
| r := require.New(t) | ||
| ctx := t.Context() | ||
|
|
||
| c, conn := adoptPair(t, ctx, | ||
| example.AdaptMeter(&exampleMeter{temp: 42}), "meter", | ||
| rpc.WithMaxFrameSize(64), | ||
| ) | ||
|
|
||
| mc := &example.MeterClient{Client: c} | ||
|
|
||
| res, err := mc.ReadTemperature(ctx, "a-meter-name-long-enough-to-need-splitting-across-frames") | ||
| r.NoError(err) | ||
| r.Equal(float32(42), res.Reading().Temperature()) | ||
|
|
||
| r.NotZero(conn.count()) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
The frame-splitting subtest doesn't actually assert splitting.
r.NotZero(conn.count()) passes with a single envelope, so this would still succeed if WithMaxFrameSize(64) were ignored. Assert a count consistent with the payload being chunked (e.g. more envelopes than an unsplit round trip would need) to make the test detect a regression.
🧪 Suggested tightening
r.Equal(float32(42), res.Reading().Temperature())
- r.NotZero(conn.count())
+ // A 64-byte cap forces the request/response payloads across several
+ // envelopes; a single frame each way would mean the cap was ignored.
+ r.Greater(conn.count(), int64(4))
})📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // A frame cap below the payload size forces msgmux to split a write across | |
| // several messages, which is what an envelope with its own size limit needs. | |
| t.Run("respects a frame size below the payload size", func(t *testing.T) { | |
| r := require.New(t) | |
| ctx := t.Context() | |
| c, conn := adoptPair(t, ctx, | |
| example.AdaptMeter(&exampleMeter{temp: 42}), "meter", | |
| rpc.WithMaxFrameSize(64), | |
| ) | |
| mc := &example.MeterClient{Client: c} | |
| res, err := mc.ReadTemperature(ctx, "a-meter-name-long-enough-to-need-splitting-across-frames") | |
| r.NoError(err) | |
| r.Equal(float32(42), res.Reading().Temperature()) | |
| r.NotZero(conn.count()) | |
| // A frame cap below the payload size forces msgmux to split a write across | |
| // several messages, which is what an envelope with its own size limit needs. | |
| t.Run("respects a frame size below the payload size", func(t *testing.T) { | |
| r := require.New(t) | |
| ctx := t.Context() | |
| c, conn := adoptPair(t, ctx, | |
| example.AdaptMeter(&exampleMeter{temp: 42}), "meter", | |
| rpc.WithMaxFrameSize(64), | |
| ) | |
| mc := &example.MeterClient{Client: c} | |
| res, err := mc.ReadTemperature(ctx, "a-meter-name-long-enough-to-need-splitting-across-frames") | |
| r.NoError(err) | |
| r.Equal(float32(42), res.Reading().Temperature()) | |
| // A 64-byte cap forces the request/response payloads across several | |
| // envelopes; a single frame each way would mean the cap was ignored. | |
| r.Greater(conn.count(), int64(4)) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/rpc/message_adopt_test.go` around lines 396 - 413, Strengthen the
“respects a frame size below the payload size” subtest by asserting that
conn.count() exceeds the envelope count expected for an unsplit round trip,
rather than only checking it is nonzero. Keep the existing temperature
assertions and use the connection’s count to verify WithMaxFrameSize(64)
actually causes payload splitting.
| // setupTCPTransport wires a client for a "tcp://host:port" remote. | ||
| func (c *NetworkClient) setupTCPTransport() { | ||
| remote := strings.TrimPrefix(c.remote, "tcp://") | ||
|
|
||
| tlsCfg := c.tlsCfg | ||
| if tlsCfg == nil { | ||
| tlsCfg = c.State.clientTlsCfg | ||
| } | ||
|
|
||
| cfg := tlsCfg.Clone() | ||
| cfg.NextProtos = nil | ||
|
|
||
| c.ops = &msgOpTransport{ | ||
| owner: c, | ||
| dial: func() (MessageConn, error) { | ||
| conn, err := tls.Dial("tcp", remote, cfg) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return newTCPConn(conn, 0), nil | ||
| }, | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
setupTCPTransport dials a fresh connection per controlSession miss without any timeout. tls.Dial blocks indefinitely on an unresponsive peer, and the dial happens under msgOpTransport.mu, so one hung dial stalls every other operation on this client. Using tls.Dialer with the caller's context (or a bounded dial timeout) would bound that.
⏱️ Proposed change
c.ops = &msgOpTransport{
owner: c,
dial: func() (MessageConn, error) {
- conn, err := tls.Dial("tcp", remote, cfg)
+ d := &tls.Dialer{Config: cfg}
+ ctx, cancel := context.WithTimeout(c.State.top, 30*time.Second)
+ defer cancel()
+ conn, err := d.DialContext(ctx, "tcp", remote)
if err != nil {
return nil, err
}
return newTCPConn(conn, 0), nil
},
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // setupTCPTransport wires a client for a "tcp://host:port" remote. | |
| func (c *NetworkClient) setupTCPTransport() { | |
| remote := strings.TrimPrefix(c.remote, "tcp://") | |
| tlsCfg := c.tlsCfg | |
| if tlsCfg == nil { | |
| tlsCfg = c.State.clientTlsCfg | |
| } | |
| cfg := tlsCfg.Clone() | |
| cfg.NextProtos = nil | |
| c.ops = &msgOpTransport{ | |
| owner: c, | |
| dial: func() (MessageConn, error) { | |
| conn, err := tls.Dial("tcp", remote, cfg) | |
| if err != nil { | |
| return nil, err | |
| } | |
| return newTCPConn(conn, 0), nil | |
| }, | |
| } | |
| } | |
| // setupTCPTransport wires a client for a "tcp://host:port" remote. | |
| func (c *NetworkClient) setupTCPTransport() { | |
| remote := strings.TrimPrefix(c.remote, "tcp://") | |
| tlsCfg := c.tlsCfg | |
| if tlsCfg == nil { | |
| tlsCfg = c.State.clientTlsCfg | |
| } | |
| cfg := tlsCfg.Clone() | |
| cfg.NextProtos = nil | |
| c.ops = &msgOpTransport{ | |
| owner: c, | |
| dial: func() (MessageConn, error) { | |
| d := &tls.Dialer{Config: cfg} | |
| ctx, cancel := context.WithTimeout(c.State.top, 30*time.Second) | |
| defer cancel() | |
| conn, err := d.DialContext(ctx, "tcp", remote) | |
| if err != nil { | |
| return nil, err | |
| } | |
| return newTCPConn(conn, 0), nil | |
| }, | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/rpc/message_tcp.go` around lines 136 - 158, Update
NetworkClient.setupTCPTransport so its msgOpTransport dial callback accepts and
propagates the caller’s context to a context-aware TLS dial, or otherwise
applies a bounded dial timeout. Preserve the existing TLS configuration and
newTCPConn setup while ensuring an unresponsive peer cannot block indefinitely.
| var vals []float32 | ||
| recv := stream.StreamRecv(func(val float32) error { | ||
| vals = append(vals, val) | ||
| return nil | ||
| }) | ||
|
|
||
| _, err := mc.Emit(ctx, recv) | ||
| r.NoError(err) | ||
|
|
||
| time.Sleep(time.Second) | ||
| r.Equal([]float32{42, 100}, vals) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Stream accumulators in pkg/rpc/transport_mem_test.go are missing the mutex the sibling tests use. Both subtests append to vals from the transport's callback goroutine and read it from the test goroutine, so go test -race will report a race; the shared root cause is the absent accumulator lock that pkg/rpc/message_net_test.go and pkg/rpc/message_adopt_test.go do apply.
pkg/rpc/transport_mem_test.go#L99-L109: guard thevalsappend insidestream.StreamRecvand ther.Equalread with async.Mutex.pkg/rpc/transport_mem_test.go#L139-L154: guard the per-goroutinevalswith its own mutex and copy it (append([]float32(nil), vals...)) intogot[i]instead of aliasing the live slice.
📍 Affects 1 file
pkg/rpc/transport_mem_test.go#L99-L109(this comment)pkg/rpc/transport_mem_test.go#L139-L154
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/rpc/transport_mem_test.go` around lines 99 - 109, Add mutex protection to
both stream accumulator tests in pkg/rpc/transport_mem_test.go: at lines 99-109,
guard vals appends in stream.StreamRecv and the r.Equal read with a sync.Mutex;
at lines 139-154, give each goroutine’s vals its own mutex, guard appends and
reads, and store a copied slice in got[i] rather than aliasing the live
accumulator.
| func (t *msgOpTransport) controlSession(ctx context.Context) (rpcSession, error) { | ||
| t.mu.Lock() | ||
| defer t.mu.Unlock() | ||
|
|
||
| if t.ctrlSess != nil { | ||
| return t.ctrlSess, nil | ||
| } | ||
| if t.dial == nil { | ||
| return nil, errors.New("rpc: message session closed and not redialable") | ||
| } | ||
| conn, err := t.dial() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| t.ctrlSess = newMsgSession(conn, true, 0) | ||
| // A connection this transport dialed lives as long as the State does. | ||
| t.startAcceptLoop(t.owner.State.top, t.ctrlSess) | ||
| return t.ctrlSess, nil | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
A dead ctrlSess is never cleared, so a dialable transport can never reconnect. controlSession caches the session for the lifetime of the transport and only consults dial when ctrlSess == nil. Once the underlying connection drops (peer restart, TCP reset, WS close), the accept loop exits but ctrlSess stays non-nil, so every subsequent OpenStreamSync fails permanently even though dial is available. Clear ctrlSess when the session terminates (e.g. in the accept-loop goroutine, guarded so an adopted session — dial == nil — stays pinned so callers still get the "not redialable" error).
🔁 Sketch
func (t *msgOpTransport) startAcceptLoop(ctx context.Context, sess rpcSession) {
st := t.owner.State
router := &sessionRouter{state: st, server: st.server, inline: &t.inline}
- go router.run(ctx, sess) //nolint:errcheck // loop ends with the session
+ go func() {
+ _ = router.run(ctx, sess) // loop ends with the session
+
+ // The session is finished; drop it so a dialable transport redials.
+ t.mu.Lock()
+ if t.ctrlSess == sess && t.dial != nil {
+ t.ctrlSess = nil
+ }
+ t.mu.Unlock()
+ }()
}Note roundTrip would also need to retry once on a stream-open failure for this to be transparent to callers.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/rpc/transport_msg.go` around lines 38 - 56, Update the
session-termination handling around controlSession and its accept-loop goroutine
to clear t.ctrlSess when a dialed control session ends, while guarding the reset
so adopted sessions with dial == nil remain pinned and continue returning the
existing “not redialable” error. Ensure the cleared session can be redialed on
the next controlSession call.
| func (t *msgOpTransport) roundTrip(ctx context.Context, req opRequest, payload any) (opReply, *cbor.Decoder, func(), error) { | ||
| sess, err := t.controlSession(ctx) | ||
| if err != nil { | ||
| return opReply{}, nil, nil, err | ||
| } | ||
|
|
||
| st, err := sess.OpenStreamSync(ctx) | ||
| if err != nil { | ||
| return opReply{}, nil, nil, err | ||
| } | ||
| done := func() { _ = st.Close() } | ||
|
|
||
| req.Version = currentProtocolVersion | ||
|
|
||
| enc := cbor.NewEncoder(st) | ||
| if err := enc.Encode(req); err != nil { | ||
| done() | ||
| return opReply{}, nil, nil, err | ||
| } | ||
| if payload != nil { | ||
| if err := enc.Encode(payload); err != nil { | ||
| done() | ||
| return opReply{}, nil, nil, err | ||
| } | ||
| } | ||
|
|
||
| dec := cbor.NewDecoder(st) | ||
| var reply opReply | ||
| if err := dec.Decode(&reply); err != nil { | ||
| done() | ||
| return opReply{}, nil, nil, err | ||
| } | ||
|
|
||
| return reply, dec, done, nil | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
roundTrip ignores ctx for the write/read phase. Cancellation only affects OpenStreamSync; if the peer never replies, dec.Decode(&reply) blocks until the session dies, so a cancelled caller's msgUnaryCall hangs. The callstream path handles this by spawning a goroutine that calls CancelRead on ctx.Done() (pkg/rpc/client.go Lines 788-791) — the same guard is missing here.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/rpc/transport_msg.go` around lines 83 - 117, Update
msgOpTransport.roundTrip to honor ctx cancellation during request encoding and
reply decoding, not only in OpenStreamSync. Add the same CancelRead-on-ctx.Done
guard used by the callstream path, ensure the watcher is cleaned up with the
stream, and return the resulting decode error when cancellation interrupts a
blocked peer response.
| func (c *NetworkClient) msgUnaryCall(ctx context.Context, method string, args, result any) error { | ||
| for { | ||
| ts := nowStamp() | ||
| req := opRequest{ | ||
| Op: opCall, | ||
| OID: c.oid, | ||
| Method: method, | ||
| Timestamp: ts, | ||
| Signature: signBytes(c.State.privkey, msgCanonical(opCall, string(c.oid)+"/"+method, ts)), | ||
| Bearer: c.bearer(), | ||
| Trace: traceMap(ctx), | ||
| } | ||
| reply, dec, done, err := c.ops.roundTrip(ctx, req, args) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| switch reply.Status { | ||
| case "ok": | ||
| err := dec.Decode(result) | ||
| done() | ||
| return err | ||
| case "unknown-capability": | ||
| done() | ||
| if c.capa != nil && c.capa.RestoreState != nil { | ||
| if rerr := c.msgReresolveCapability(c.capa.RestoreState); rerr == nil { | ||
| continue | ||
| } | ||
| } | ||
| return cond.NotFound("capability", c.oid) | ||
| case "error": | ||
| done() | ||
| return cond.RemoteError(reply.Category, reply.Code, reply.Error) | ||
| case "forbidden": | ||
| done() | ||
| // An authorization denial is a permission error, not a protocol one; | ||
| // without this case it would surface as "unknown response status". | ||
| return cond.RemoteError("permission", "forbidden", reply.Error) | ||
| case "panic": | ||
| done() | ||
| return cond.Panic(reply.Error) | ||
| default: | ||
| done() | ||
| return fmt.Errorf("unknown response status to %s/%s: %s", c.oid, method, reply.Status) | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Unbounded re-resolve loop. If the server keeps answering unknown-capability while msgReresolveCapability keeps succeeding, this for loop spins issuing RPCs forever. Bounding it to a single retry matches the intent and mirrors the HTTP path's practical behavior.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/rpc/transport_msg.go` around lines 319 - 365, The msgUnaryCall retry loop
currently re-resolves capabilities and retries indefinitely when the server
repeatedly returns unknown-capability. Limit this recovery path to one
re-resolve and retry attempt, then return cond.NotFound if unknown-capability
persists or re-resolution fails, while preserving normal handling for all other
reply statuses.
| func isRetryableTransportError(err error) bool { | ||
| var ae *quic.ApplicationError | ||
| return errors.As(err, &ae) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Inspect quic.ApplicationError shape and how retries are bounded.
rg -nP -C4 'isRetryableTransportError' pkg/rpc
rg -nP -C3 'continue request' pkg/rpc/client.goRepository: mirendev/runtime
Length of output: 2262
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the retry loop and the QUIC transport error handling.
sed -n '560,780p' pkg/rpc/client.go
printf '\n----\n'
sed -n '1,140p' pkg/rpc/transport_wt.go
# Find any retry counters / backoff / caps in rpc client code.
rg -n 'retry|backoff|attempt|limit|max|continue request' pkg/rpc/client.go pkg/rpc/transport_wt.goRepository: mirendev/runtime
Length of output: 9672
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Find the QUIC error type in vendored/module sources if present in the repo tree.
rg -n 'type ApplicationError|ApplicationError struct|ErrorCode' .Repository: mirendev/runtime
Length of output: 2335
🌐 Web query:
quic-go ApplicationError errors.As application close error code retry semantics
💡 Result:
In quic-go, an ApplicationError is an application-defined error that occurs when the remote peer initiates a connection close with an application-specific error code [1][2][3]. To handle and inspect these errors, you use errors.As to assert the error to *quic.ApplicationError [4]. This provides access to the ErrorCode, ErrorMessage, and a Remote boolean indicating if the error was triggered by the peer [2][4]. Retry semantics for ApplicationError are not automatically managed by the transport layer; they are application-dependent [4][5]. Whether it is safe to retry depends on the specific error code and the state of the operation [5]. For example, in HTTP/3, quic-go logic distinguishes between operations that are safe to retry (e.g., if the stream was never successfully sent or if a specific rejection code like H3_REQUEST_REJECTED is received) and those that are not [5]. When implementing retries: 1. Use errors.As to extract the *quic.ApplicationError [4]. 2. Check the ErrorCode against your application's defined protocol codes [2][4]. 3. Only retry if the operation is idempotent or if you can confirm that no request data (or partial/invalid state) was processed by the peer [5]. Example usage: if appErr, ok:= errors.As(err, new(*quic.ApplicationError)); ok { // Inspect appErr.ErrorCode and decide on retry logic log.Printf("Application error code: %x, remote: %t", appErr.ErrorCode, appErr.Remote) }
Citations:
- 1: https://github.com/quic-go/quic-go/blob/master/errors.go
- 2: https://github.com/quic-go/quic-go/blob/70764cde/internal/qerr/errors.go
- 3: https://github.com/quic-go/quic-go/blob/master/connection.go
- 4: https://quic-go.net/docs/quic/connection/
- 5: http3: only retry requests for which it is safe to do so quic-go/quic-go#5141
Treat only the intended QUIC application-error codes as retryable pkg/rpc/transport_wt.go:66-68
errors.As(err, &ae) currently treats every *quic.ApplicationError as retryable, and both Call and handleCallStreamDialError loop back on that signal. A server that keeps closing with an application error can drive the client into an endless retry loop. Narrow this to the specific retryable codes and/or add a retry limit/backoff.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/rpc/transport_wt.go` around lines 66 - 69, Update
isRetryableTransportError to return true only for the explicitly intended
retryable QUIC application-error codes, rather than every
*quic.ApplicationError; preserve non-application-error handling and ensure Call
and handleCallStreamDialError stop retrying for other codes.
1 participant
Our RPC framework could only run over one thing: QUIC/HTTP3 with WebTransport. If UDP/QUIC was blocked, or you had a connection that wasn't QUIC, you were out of luck. It couldn't fall back to TCP, and it certainly couldn't run over a socket it didn't stand up itself. That walls us off from two things we want to build.
The one driving this: we want miren.cloud to be able to make RPC calls into a cluster. That connection is a WebSocket that cloud owns, carrying cloud's own envelope protocol, and rpc will be handed already-unwrapped payloads rather than getting to dial, listen, or close anything. The framework had no seam for "here's a pipe, run RPC over it." The second, smaller need is TCP clients: today there's no non-QUIC path to the cluster API at all, and we want a lean one.
Both come down to the same thing: run RPC over a single, externally supplied, message-oriented pipe. So this adds
MessageConn, an interface that's justSend,Recv,Closeover discrete byte messages, with a small stream multiplexer (msgmux) layered on top to recover the callback and streaming semantics the framework needs.ServeMessageConnandClientFromMessageConnare the public entry points. The cloud's enveloped socket becomes just another backend behind that interface, andws://andtcp://fall out as the first non-QUIC transports for free.The interesting hard part is that RPC here is bidirectional: a client advertises a capability and the server calls back on it over the same connection. WebTransport gets that for free because either QUIC peer can open a substream anytime. A single pipe can't, so every stream on a message session opens with a small header, and one accept loop routes the whole session by that header. This is also what lets a cluster that dialed outbound through a firewall still be called back into, which is exactly the cloud topology.
Two supporting changes came along for the ride.
Authenticatorused to take an*http.Request, which meant none of our real authenticators (cloudauth, oidcauth, outboard) could run on a transport that has no HTTP request. It now takes a transport-neutralCredentials, and a bearer token rides in the message header where there's no place to put one. And there's aPROTOCOL.mdplus a version field on the wire, so a second (non-Go) implementation later won't force a breaking change.To be clear about scope: this is groundwork for building out the ability to make RPC calls to a cluster from miren.cloud. It builds the transport and proves it end to end (there's a test that runs RPC over a caller-owned enveloped pipe with no dialer at all, the closest proxy for the cloud case), but wiring it into an actual miren.cloud to cluster call path is follow-up work. This is the foundation that makes it possible.
Along the way it also fixes a pre-existing data race on
heldCapability.lastContactthat concurrent calls against one capability were all writing unsynchronized.