diff --git a/doc/dev-guide.md b/doc/dev-guide.md index fd302f87..f3d6fe0a 100644 --- a/doc/dev-guide.md +++ b/doc/dev-guide.md @@ -165,6 +165,9 @@ It holds a `channelRef`, an atomically replaceable reference to the node's curre There are three channel kinds: * `OutboundChannel` runs over streams this node dials; it opens a new `session` for each stream and requeues pending calls when a stream is lost, as far as the send queue has space. + A stream that the server refuses is not lost: it ends with a refusal code (`Unauthenticated`, `PermissionDenied`, `InvalidArgument`, or `Unimplemented`) before it delivered a frame. + Its pending calls fail with the refusal, and the channel backs off before it opens the next stream, failing requests at once with the refusal until then. + A stream opened after a refusal is suspect: it counts as up only once it delivers a frame or lives 2 s, and until then its sends do not clear `LastErr`. * `InboundChannel` runs over one stream that the server accepted; its single `session` ends with the stream. * `LocalChannel` serves requests in-process through the handler, with no stream. @@ -173,7 +176,7 @@ A `session` sends queued requests on its `BidiStream`, records two-way calls in The dispatcher runs handlers one at a time, in arrival order. A handler that returns without calling `Release` hands its goroutine to the next queued handler, so a busy stream keeps one goroutine and the stack it has grown; an early `Release` starts the next handler on a new goroutine. On the server side, `Server.NodeStream` asks a `PeerAcceptor` for the `InboundChannel` of each accepted stream. -An `OutboundChannel` with a `PeerVerifier` checks the remote end of each new stream against the channel's node ID, with the stream's context, before it sends requests on the stream; a stream that fails the check is closed, and the waiting request fails with `codes.Unauthenticated`. +An `OutboundChannel` with a `PeerVerifier` checks the remote end of each new stream against the channel's node ID, with the stream's context, before it sends requests on the stream; a stream that fails the check is closed, the waiting request fails with `codes.Unauthenticated`, and the channel backs off as for a refused stream. ```mermaid classDiagram @@ -242,6 +245,28 @@ classDiagram PeerAcceptor ..> InboundChannel : returns ``` +Only the receive loop of an outbound session learns whether the server refused its stream: when the server ends a stream, a concurrent `Send` returns `io.EOF`, and only `Recv` returns the server's status. +So a send that fails with `io.EOF` before the stream delivered a frame only cancels the stream, and the receive loop disposes of the pending calls. +The session's `handoff` state ensures that exactly one goroutine disposes of them. +The session's `verdict` channel closes when the stream delivers its first frame or receiving ends; the run loop waits for it after a session ends, and then reads whether the server refused the stream. + +```mermaid +stateDiagram-v2 + [*] --> Fresh: stream opened + Fresh --> Accepted: first frame arrives + Fresh --> SendFailed: a send fails with io.EOF + Fresh --> Closed: Recv fails + SendFailed --> Accepted: a buffered frame arrives, so retry the pending calls + SendFailed --> Closed: Recv fails + Accepted --> Closed: Recv fails + Closed --> [*] + note right of Closed + From Fresh or SendFailed with a refusal code: + fail the pending calls with the refusal. + Otherwise: retry them. + end note +``` + ### Connectivity Layer (`internal/conn`) A `Config` is a slice of `*Node`, and each `Node` has a fixed `stream.Transport`. diff --git a/doc/user-guide.md b/doc/user-guide.md index 6dc813af..c05be6d1 100644 --- a/doc/user-guide.md +++ b/doc/user-guide.md @@ -2404,12 +2404,20 @@ Use `TLSPeerVerifier` to bind node IDs to mTLS certificates; see [Binding Node I | --- | --- | --- | | Claims the node ID of a configured peer | `verify(ctx, claimed ID)` before the stream is attached to the peer | The stream ends with `codes.Unauthenticated`; no connect callback or handler runs | | Claims a node ID outside the peer configuration | None; there is no identity to check | The stream ends with `codes.Unauthenticated` | -| Opened by this server to a configured peer | `verify(ctx, dialed ID)` before any request is sent | The waiting call fails with `codes.Unauthenticated`; the server tries again later | +| Opened by this server to a configured peer | `verify(ctx, dialed ID)` before any request is sent | The waiting call fails with `codes.Unauthenticated`; the server backs off before it opens a new stream to the peer, as a refused peer does | | Claims no peer node ID (a client) | None | Served as without the option | -A peer that this server refuses sees the refusal, with the verifier's text, in `Node.LastErr` of its node for this server. -Its pending two-way calls to this server are resent on new streams until their contexts end, so they fail with the context's error, not with `codes.Unauthenticated`; give calls between peers a deadline. -While the refused peer keeps retrying, its `ConnectedPeers` view can briefly include this server, and under `WithStreamDedup` its `WaitForAll` can return before this server has accepted any of its streams. +A peer that this server refuses sees the refusal, with the verifier's text, in `Node.LastErr` of its node for this server, until one of its streams to this server succeeds. +Its calls on the refused stream fail with the refusal's status; they are not resent. +A node treats a stream as refused when the stream ends with `codes.Unauthenticated`, `codes.PermissionDenied`, `codes.InvalidArgument`, or `codes.Unimplemented` before it delivered any message. +After a refusal, the refused peer backs off before it opens a new stream to this server: 50 ms after the first refusal, doubling after each further refusal up to 2 s. +Until the backoff ends, its calls to this server fail at once with the refusal. +The refused peer's first stream counts as connected until the refusal arrives, about one round trip later. +After a refusal, a new stream counts as connected only once it delivers a message or has lived 2 s. +After the refusal of its first stream arrives, its `ConnectedPeers` view no longer includes this server while this server keeps refusing it, and under `WithStreamDedup` its `WaitForAll` does not return. +A `WaitForAll` that is already waiting when the first stream opens can still return during that round trip. +Once this server accepts the peer, an idle stream counts as connected 2 s after it opened, or earlier if a message arrives on it. +When a stream that this server opened fails this server's own check, the peer at the other end counts this server as connected until the stream is closed, about one round trip, once per backoff period. With `WithPeerVerifier`, `ServerContext.SenderID` is authenticated for every configured peer, with and without stream deduplication: diff --git a/internal/conn/node.go b/internal/conn/node.go index c719c087..61eaa83a 100644 --- a/internal/conn/node.go +++ b/internal/conn/node.go @@ -336,8 +336,10 @@ func (n *Node) Detail() string { // // It reports current node health, not the outcome of any particular call: it is // last-write-wins across concurrent requests and reverts to nil once traffic -// flows to the node again. Use the [ByLastError] comparator with [Config.Sort] -// to order nodes by whether they are currently failing. +// flows to the node again. After the node refused a stream, it reverts to nil +// only once a stream proves viable: the stream delivers a message, or a send +// succeeds on a stream that has lived 2 s. Use the [ByLastError] comparator +// with [Config.Sort] to order nodes by whether they are currently failing. func (n *Node) LastErr() error { if ch := n.activeChannel(); ch != nil { return ch.LastErr() diff --git a/internal/stream/channel.go b/internal/stream/channel.go index 0b2244f5..437c26fc 100644 --- a/internal/stream/channel.go +++ b/internal/stream/channel.go @@ -19,10 +19,14 @@ type Channel interface { StreamUp() bool // LastErr returns [ErrSendStalled] while a send has been blocked for // [stallReportDelay] or longer, and otherwise the error of the channel's - // most recent stream operation, or nil if it succeeded. + // most recent stream operation, or nil if it succeeded. After the server + // refused a stream, LastErr reports the refusal until a stream proves + // viable: it delivers a frame, or a send succeeds on a stream that has + // lived [eagerReconnectMaxDelay]. LastErr() error // DroppedReplies returns the number of handler replies dropped because - // the send queue was full or closed. + // the send queue was full or closed, or because they failed during the + // backoff after the server refused a stream. DroppedReplies() int64 // PendingCount returns the number of two-way calls awaiting responses. // The count may include calls whose callers are done but that have not diff --git a/internal/stream/channel_test.go b/internal/stream/channel_test.go index faa72f65..18839245 100644 --- a/internal/stream/channel_test.go +++ b/internal/stream/channel_test.go @@ -296,12 +296,23 @@ func waitForDisconnection(c *OutboundChannel, timeout time.Duration) bool { func sendRequest(t testing.TB, c Channel, req Request, msgID uint64) response { t.Helper() + resp, err := trySendRequest(c, req, msgID) + if err != nil { + t.Fatal(err) + } + return resp +} + +// trySendRequest is [sendRequest] without the test: it returns an error +// instead of failing the test, so that goroutines other than the test's can +// call it. +func trySendRequest(c Channel, req Request, msgID uint64) (response, error) { if req.Ctx == nil { req.Ctx = context.Background() } reqMsg, err := NewMessage(req.Ctx, msgID, mock.TestMethod, nil) if err != nil { - t.Fatalf("NewMessage failed: %v", err) + return response{}, fmt.Errorf("NewMessage failed: %w", err) } req.Msg = reqMsg responseChan := make(chan response, 1) @@ -310,10 +321,9 @@ func sendRequest(t testing.TB, c Channel, req Request, msgID uint64) response { select { case resp := <-responseChan: - return resp + return resp, nil case <-time.After(defaultTestTimeout): - t.Fatalf("timeout waiting for response to message %d", msgID) - return response{} + return response{}, fmt.Errorf("timeout waiting for response to message %d", msgID) } } @@ -1268,12 +1278,13 @@ func TestChannelEagerReconnectRedialsWithoutSends(t *testing.T) { } // rejectEveryStreamServer returns a NodeStream server function that rejects -// every accepted stream immediately and counts how many streams it accepted. -func rejectEveryStreamServer() (serverFn func(Gorums_NodeStreamServer) error, count *atomic.Int32) { +// every accepted stream immediately with err and counts how many streams it +// accepted. +func rejectEveryStreamServer(err error) (serverFn func(Gorums_NodeStreamServer) error, count *atomic.Int32) { count = new(atomic.Int32) serverFn = func(Gorums_NodeStreamServer) error { count.Add(1) - return errors.New("stream rejected by test server") + return err } return serverFn, count } @@ -1281,21 +1292,444 @@ func rejectEveryStreamServer() (serverFn func(Gorums_NodeStreamServer) error, co // TestChannelEagerReconnectBacksOffRejectedStreams verifies that a channel with // eager reconnection paces its redials with capped backoff when the server // rejects every stream, instead of spinning and creating a new stream per -// iteration. Without backoff the loop produced thousands of accepted streams -// (5,127 in 250 ms during review); with backoff (50 ms base, doubling to a 2 s -// cap) only a handful of attempts fit in the window below. +// iteration, whether the rejection is a refusal or not. Without backoff the +// loop produced thousands of accepted streams (5,127 in 250 ms during +// review); with backoff (50 ms base, doubling to a 2 s cap) only a handful of +// attempts fit in the window below. func TestChannelEagerReconnectBacksOffRejectedStreams(t *testing.T) { - serverFn, count := rejectEveryStreamServer() - setupChannelEager(t, true, serverFn) + tests := []struct { + name string + err error + }{ + {name: "Unknown", err: errors.New("stream rejected by test server")}, + {name: "Unauthenticated", err: status.Error(codes.Unauthenticated, "stream refused by test server")}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + serverFn, count := rejectEveryStreamServer(tt.err) + setupChannelEager(t, true, serverFn) + + const window = 500 * time.Millisecond + time.Sleep(window) + // Attempts within the window land at roughly 0, 50, 150, 350 ms + // plus the sender's initial eager connect: about five. The + // generous bound tolerates scheduler jitter while still catching + // an unpaced spin. + if got := count.Load(); got > 20 { + t.Fatalf("server accepted %d streams in %v; eager reconnect is not backing off (want <= 20)", got, window) + } + }) + } +} + +// refusingServer is a NodeStream server that refuses streams with code. If +// read is set, it first reads one request from a stream and then waits hold, +// so that the requests sent meanwhile are pending on the stream when it is +// refused. Once accept is set, it echoes on new streams instead. +type refusingServer struct { + code codes.Code + read bool + hold time.Duration + accept atomic.Bool + streams atomic.Int32 // the number of streams opened to the server +} + +// refuseStreamServer returns a server that refuses every stream with code +// at once. +func refuseStreamServer(code codes.Code) *refusingServer { + return &refusingServer{code: code} +} + +func (r *refusingServer) serve(stream Gorums_NodeStreamServer) error { + r.streams.Add(1) + if r.accept.Load() { + return echoServer(stream) + } + if r.read { + if _, err := stream.Recv(); err != nil { + return err + } + time.Sleep(r.hold) + } + return status.Error(r.code, "stream refused by test server") +} + +// refusalModes are the channel configurations under which a refused stream +// must behave the same. +var refusalModes = []struct { + name string + eager bool +}{ + {name: "Lazy", eager: false}, + {name: "Eager", eager: true}, +} + +// TestChannelRefusedStreamFailsCalls verifies that a call on a channel whose +// server refuses its streams fails promptly with the refusal's code, for +// two-way and streaming calls, with and without eager reconnect. +func TestChannelRefusedStreamFailsCalls(t *testing.T) { + codesUnderTest := []codes.Code{codes.Unauthenticated, codes.PermissionDenied, codes.InvalidArgument, codes.Unimplemented} + for _, code := range codesUnderTest { + for _, mode := range refusalModes { + for _, streaming := range []bool{false, true} { + t.Run(fmt.Sprintf("%v/%s/Streaming=%t", code, mode.name, streaming), func(t *testing.T) { + srv := refuseStreamServer(code) + // The server reads the call, so the call is pending on the + // refused stream unless it is made during a backoff. + srv.read = true + tc := setupChannelEager(t, mode.eager, srv.serve) + resp := sendRequest(t, tc.OutboundChannel, Request{Streaming: streaming}, 1) + if got := status.Code(resp.Err); got != code { + t.Fatalf("call failed with %v, want code %v", resp.Err, code) + } + if got := srv.streams.Load(); got > 2 { + t.Errorf("server saw %d streams, want at most 2", got) + } + }) + } + } + } +} + +// TestChannelRefusalBacksOff verifies that a channel paces the streams it +// opens to a server that refuses them, while calls keep failing with the +// refusal, with and without eager reconnect. +func TestChannelRefusalBacksOff(t *testing.T) { + for _, mode := range refusalModes { + t.Run(mode.name, func(t *testing.T) { + srv := refuseStreamServer(codes.Unauthenticated) + tc := setupChannelEager(t, mode.eager, srv.serve) + + const window = 500 * time.Millisecond + calls := 0 + for deadline := time.Now().Add(window); time.Now().Before(deadline); calls++ { + resp := sendRequest(t, tc.OutboundChannel, Request{}, uint64(calls+1)) + if got := status.Code(resp.Err); got != codes.Unauthenticated { + t.Fatalf("call %d failed with %v, want code %v", calls+1, resp.Err, codes.Unauthenticated) + } + } + // Attempts land at roughly 0, 50, 150, and 350 ms, since the + // delay doubles after each refusal. A delay that stays at 50 ms + // gives about 10 attempts, and an unpaced loop thousands. + // Scheduler jitter only delays attempts. + if got := srv.streams.Load(); got > 7 { + t.Errorf("server saw %d streams during %d calls in %v, want at most 7", got, calls, window) + } + }) + } +} + +// TestChannelRefusalFailsCallsDuringBackoff verifies that calls made right +// after a refusal fail at once with the refusal, without a new stream each. +func TestChannelRefusalFailsCallsDuringBackoff(t *testing.T) { + srv := refuseStreamServer(codes.PermissionDenied) + tc := setupChannel(t, srv.serve) + // The channel's first stream, opened at once, is refused. + if !waitUntil(defaultTestTimeout, func() bool { return status.Code(tc.LastErr()) == codes.PermissionDenied }) { + t.Fatalf("LastErr = %v, want code %v", tc.LastErr(), codes.PermissionDenied) + } + for i := range 10 { + resp := sendRequest(t, tc.OutboundChannel, Request{}, uint64(i+1)) + if got := status.Code(resp.Err); got != codes.PermissionDenied { + t.Fatalf("call %d failed with %v, want code %v", i+1, resp.Err, codes.PermissionDenied) + } + } + if got := srv.streams.Load(); got > 2 { + t.Errorf("server saw %d streams, want at most 2", got) + } +} + +// TestChannelRefusalCountsDroppedReplies verifies that a handler reply that +// fails during a refusal backoff, which has no caller to report to, counts in +// DroppedReplies. +func TestChannelRefusalCountsDroppedReplies(t *testing.T) { + srv := refuseStreamServer(codes.Unauthenticated) + tc := setupChannel(t, srv.serve) + if !waitUntil(defaultTestTimeout, func() bool { return status.Code(tc.LastErr()) == codes.Unauthenticated }) { + t.Fatalf("LastErr = %v, want code %v", tc.LastErr(), codes.Unauthenticated) + } + for i := range 10 { + tc.reply(Message_builder{MessageSeqNo: ServerSequenceNumber(uint64(i + 1)), Method: mock.TestMethod}.Build()) + } + if !waitUntil(defaultTestTimeout, func() bool { return tc.DroppedReplies() > 0 }) { + t.Error("DroppedReplies() = 0 after replies failed during a refusal backoff, want > 0") + } +} + +// TestChannelRefusalRecovers verifies that a channel whose server refused its +// streams carries calls again once the server accepts them, with and without +// eager reconnect. +func TestChannelRefusalRecovers(t *testing.T) { + for _, mode := range refusalModes { + t.Run(mode.name, func(t *testing.T) { + srv := refuseStreamServer(codes.Unauthenticated) + srv.read = true + tc := setupChannelEager(t, mode.eager, srv.serve) + if resp := sendRequest(t, tc.OutboundChannel, Request{}, 1); status.Code(resp.Err) != codes.Unauthenticated { + t.Fatalf("call failed with %v, want code %v", resp.Err, codes.Unauthenticated) + } + srv.accept.Store(true) + var err error + msgID := uint64(1) + if !waitUntil(defaultTestTimeout, func() bool { + msgID++ + err = sendRequest(t, tc.OutboundChannel, Request{}, msgID).Err + return err == nil + }) { + t.Fatalf("call after the server accepts failed with %v, want success", err) + } + }) + } +} + +// TestChannelRefusalKeepsLastErr verifies that LastErr keeps reporting a +// refusal while the server refuses the channel's streams, even while a +// request is sent on a new stream that the server will refuse, and that it +// clears once a round trip completes after the server accepts. +func TestChannelRefusalKeepsLastErr(t *testing.T) { + for _, mode := range refusalModes { + t.Run(mode.name, func(t *testing.T) { + srv := refuseStreamServer(codes.Unauthenticated) + // The server holds each stream for a while after the request + // that it reads, before it refuses the stream. + srv.read, srv.hold = true, 20*time.Millisecond + tc := setupChannelEager(t, mode.eager, srv.serve) + if resp := sendRequest(t, tc.OutboundChannel, Request{}, 1); status.Code(resp.Err) != codes.Unauthenticated { + t.Fatalf("call failed with %v, want code %v", resp.Err, codes.Unauthenticated) + } + + const window = 600 * time.Millisecond + ctx, stop := context.WithTimeout(t.Context(), window) + defer stop() + var wg sync.WaitGroup + wg.Go(func() { + for msgID := uint64(2); ctx.Err() == nil; msgID++ { + if _, err := trySendRequest(tc.OutboundChannel, Request{}, msgID); err != nil { + t.Error(err) + return + } + } + }) + for ctx.Err() == nil { + if err := tc.LastErr(); status.Code(err) != codes.Unauthenticated { + t.Errorf("LastErr = %v while the server refuses every stream, want code %v", err, codes.Unauthenticated) + break + } + time.Sleep(100 * time.Microsecond) + } + wg.Wait() + + srv.accept.Store(true) + msgID := uint64(1 << 20) + if !waitUntil(defaultTestTimeout, func() bool { + msgID++ + return sendRequest(t, tc.OutboundChannel, Request{}, msgID).Err == nil + }) { + t.Fatal("no call succeeded after the server accepts") + } + if err := tc.LastErr(); err != nil { + t.Errorf("LastErr = %v after a round trip on an accepted stream, want nil", err) + } + }) + } +} + +// TestChannelStreamUpAfterRefusal verifies that, after a refusal, a new +// stream counts as up only once it proves viable: streams that the server +// refuses later never call OnStreamChange, and a stream that the server +// accepts comes up at its first frame. +func TestChannelStreamUpAfterRefusal(t *testing.T) { + for _, mode := range refusalModes { + t.Run(mode.name, func(t *testing.T) { + srv := refuseStreamServer(codes.Unauthenticated) + srv.read, srv.hold = true, 5*time.Millisecond + var changes atomic.Int32 + opts := OutboundOptions{EagerReconnect: mode.eager, OnStreamChange: func() { changes.Add(1) }} + tc := setupChannelOptions(t, opts, srv.serve) + + // The first stream comes up at once and goes down at its refusal. + if resp := sendRequest(t, tc.OutboundChannel, Request{}, 1); status.Code(resp.Err) != codes.Unauthenticated { + t.Fatalf("call failed with %v, want code %v", resp.Err, codes.Unauthenticated) + } + if !waitUntil(defaultTestTimeout, func() bool { return changes.Load() == 2 }) { + t.Fatalf("OnStreamChange called %d times for the first stream, want 2", changes.Load()) + } + + // Later attempts, each with a call, are refused without coming up. + for deadline, msgID := time.Now().Add(400*time.Millisecond), uint64(2); time.Now().Before(deadline); msgID++ { + if resp := sendRequest(t, tc.OutboundChannel, Request{}, msgID); status.Code(resp.Err) != codes.Unauthenticated { + t.Fatalf("call failed with %v, want code %v", resp.Err, codes.Unauthenticated) + } + } + if got := srv.streams.Load(); got < 3 { + t.Fatalf("server saw %d streams, want at least 3", got) + } + if got := changes.Load(); got != 2 { + t.Errorf("OnStreamChange called %d times, want 2: a refused stream came up", got) + } + + // A stream that the server accepts comes up at its first frame, + // well before the probation of an idle stream ends. + srv.accept.Store(true) + msgID := uint64(1 << 20) + if !waitUntil(defaultTestTimeout, func() bool { + msgID++ + return sendRequest(t, tc.OutboundChannel, Request{}, msgID).Err == nil + }) { + t.Fatal("no call succeeded after the server accepts") + } + if !waitUntil(eagerReconnectMaxDelay/4, tc.StreamUp) { + t.Error("StreamUp = false after a round trip on an accepted stream, want true") + } + }) + } +} + +// TestChannelRefusalRacesSends verifies that every call fails with the +// refusal when many calls race with the refusal of a stream: the calls +// pending on the stream, a call whose send fails as the stream ends, and the +// calls queued behind them. Two-way and streaming calls alternate. It is the +// regression test for the handoff of pending calls from a failed send to the +// receive loop, and is most useful under the race detector. +func TestChannelRefusalRacesSends(t *testing.T) { + for _, mode := range refusalModes { + t.Run(mode.name, func(t *testing.T) { + srv := refuseStreamServer(codes.Unauthenticated) + srv.read = true + tc := setupChannelOptions(t, OutboundOptions{SendBufferSize: 64, EagerReconnect: mode.eager}, srv.serve) + + const window = 500 * time.Millisecond + ctx, stop := context.WithTimeout(t.Context(), window) + defer stop() + var msgIDs atomic.Uint64 + var wg sync.WaitGroup + // One-way calls keep the send loop busy, so that sends race with + // each refusal; their outcomes vary with that race. + for range 8 { + wg.Go(func() { + for ctx.Err() == nil { + if _, err := trySendRequest(tc.OutboundChannel, Request{Oneway: true}, msgIDs.Add(1)); err != nil { + t.Error(err) + return + } + } + }) + } + // Two-way and streaming calls must fail with the refusal. + for range 16 { + wg.Go(func() { + for i := 0; ctx.Err() == nil; i++ { + msgID := msgIDs.Add(1) + resp, err := trySendRequest(tc.OutboundChannel, Request{Streaming: i%2 == 1}, msgID) + if err != nil { + t.Error(err) + return + } + if got := status.Code(resp.Err); got != codes.Unauthenticated { + t.Errorf("call %d (streaming %t) failed with %v, want code %v", msgID, i%2 == 1, resp.Err, codes.Unauthenticated) + return + } + } + }) + } + wg.Wait() + if got := srv.streams.Load(); got > 20 { + t.Errorf("server saw %d streams in %v, want at most 20", got, window) + } + }) + } +} + +// TestChannelVerifyFailureBacksOff verifies that a channel whose own check of +// the server fails paces the streams it opens, while calls keep failing with +// codes.Unauthenticated. +func TestChannelVerifyFailureBacksOff(t *testing.T) { + var streams atomic.Int32 + countingEcho := func(stream Gorums_NodeStreamServer) error { + streams.Add(1) + return echoServer(stream) + } + opts := OutboundOptions{Verify: func(context.Context, ID) error { return errors.New("certificate names another node") }} + tc := setupChannelOptions(t, opts, countingEcho) const window = 500 * time.Millisecond - time.Sleep(window) - // Attempts within the window land at roughly 0, 50, 150, 350 ms plus the - // sender's initial eager connect: about five. The generous bound tolerates - // scheduler jitter while still catching an unpaced spin. - if got := count.Load(); got > 20 { - t.Fatalf("server accepted %d streams in %v; eager reconnect is not backing off (want <= 20)", got, window) + calls := 0 + for deadline := time.Now().Add(window); time.Now().Before(deadline); calls++ { + resp := sendRequest(t, tc.OutboundChannel, Request{}, uint64(calls+1)) + if got := status.Code(resp.Err); got != codes.Unauthenticated { + t.Fatalf("call %d failed with %v, want code %v", calls+1, resp.Err, codes.Unauthenticated) + } + } + // As in TestChannelRefusalBacksOff, the doubling delay allows about five + // attempts in the window. + if got := streams.Load(); got > 7 { + t.Errorf("server saw %d streams during %d calls in %v, want at most 7", got, calls, window) + } +} + +// TestChannelLostStreamBeforeFrameRetries verifies that a call pending on a +// stream that ends with a status other than a refusal, before the stream +// delivered a frame, is sent again on a new stream. +func TestChannelLostStreamBeforeFrameRetries(t *testing.T) { + for _, code := range []codes.Code{codes.Unavailable, codes.Unknown, codes.Internal} { + t.Run(code.String(), func(t *testing.T) { + var streams atomic.Int32 + serverFn := func(stream Gorums_NodeStreamServer) error { + if streams.Add(1) == 1 { + if _, err := stream.Recv(); err != nil { + return err + } + return status.Error(code, "stream lost by test server") + } + return echoServer(stream) + } + tc := setupChannel(t, serverFn) + if resp := sendRequest(t, tc.OutboundChannel, Request{}, 1); resp.Err != nil { + t.Fatalf("call failed with %v, want it resent and answered", resp.Err) + } + }) + } +} + +// TestChannelRefusalAfterFrameRetries verifies that a stream that delivered a +// frame and then ends with a refusal code counts as lost, not refused: its +// pending call is sent again on a new stream. +func TestChannelRefusalAfterFrameRetries(t *testing.T) { + var streams atomic.Int32 + serverFn := func(stream Gorums_NodeStreamServer) error { + if streams.Add(1) > 1 { + return echoServer(stream) + } + in, err := stream.Recv() + if err != nil { + return err + } + if err := stream.Send(in); err != nil { + return err + } + if _, err := stream.Recv(); err != nil { + return err + } + return status.Error(codes.Unauthenticated, "stream ended by test server") + } + tc := setupChannel(t, serverFn) + if resp := sendRequest(t, tc.OutboundChannel, Request{}, 1); resp.Err != nil { + t.Fatalf("first call failed with %v", resp.Err) + } + if resp := sendRequest(t, tc.OutboundChannel, Request{}, 2); resp.Err != nil { + t.Fatalf("call pending on the ended stream failed with %v, want it resent and answered", resp.Err) + } +} + +// waitUntil polls cond every millisecond until it holds or timeout expires, +// and reports whether it held. +func waitUntil(timeout time.Duration, cond func() bool) bool { + for deadline := time.Now().Add(timeout); time.Now().Before(deadline); time.Sleep(time.Millisecond) { + if cond() { + return true + } } + return cond() } type signalingRequestHandler struct { diff --git a/internal/stream/errors.go b/internal/stream/errors.go index 0fd05c5e..353a5163 100644 --- a/internal/stream/errors.go +++ b/internal/stream/errors.go @@ -33,3 +33,16 @@ const stallReportDelay = time.Second func VerificationError(id ID, err error) error { return status.Errorf(codes.Unauthenticated, "gorums: node %d failed verification: %v", id, err) } + +// isRefusal reports whether err has a status code with which the remote end +// refuses a stream, so that a new stream to the same server fails the same +// way: the server does not accept this node's identity (Unauthenticated, +// PermissionDenied), rejects its stream (InvalidArgument), or does not serve +// Gorums streams (Unimplemented). +func isRefusal(err error) bool { + switch status.Code(err) { + case codes.Unauthenticated, codes.PermissionDenied, codes.InvalidArgument, codes.Unimplemented: + return true + } + return false +} diff --git a/internal/stream/outbound.go b/internal/stream/outbound.go index 759c04e6..14e4fc24 100644 --- a/internal/stream/outbound.go +++ b/internal/stream/outbound.go @@ -37,7 +37,9 @@ type OutboundOptions struct { // Latency receives round-trip samples; may be nil. Latency *Latency // EagerReconnect makes the channel re-establish a lost stream with capped - // exponential backoff, instead of on the next request. + // exponential backoff, instead of on the next request. After a refused + // stream, it makes the channel open a new stream when the backoff ends, + // instead of on the first request after that. EagerReconnect bool // OnStreamChange, if non-nil, is called on the channel's goroutine each // time the stream comes up or goes down; [OutboundChannel.StreamUp] @@ -63,9 +65,22 @@ type OutboundOptions struct { // stream. Calls already sent on the old stream complete there, and the old // stream ends once it has no calls left. // +// A server refuses a stream when the stream ends with codes.Unauthenticated, +// codes.PermissionDenied, codes.InvalidArgument, or codes.Unimplemented +// before it delivered a frame. Calls on a refused stream fail with the +// refusal's status and are not sent again. The channel then backs off before +// it opens a new stream: 50 ms after the first refusal, doubling after each +// further refusal up to 2 s, and starting over once a stream delivers a +// frame or lives 2 s. Until the backoff ends, each request fails at once +// with the refusal. After a refusal, a new stream counts as up (see +// [OutboundChannel.StreamUp]) only once it delivers a frame or lives 2 s, so +// a stream that the server refuses again never counts as up. The first +// refused stream counts as up until its refusal arrives. +// // With [OutboundOptions.Verify], a stream whose remote end fails the check is -// closed before any request is sent on it, and the waiting request fails with -// the error from [VerificationError]. +// closed before any request is sent on it, the waiting request fails with +// the error from [VerificationError], and the channel backs off as for a +// refused stream. type OutboundChannel struct { endpoint conn *grpc.ClientConn @@ -108,11 +123,19 @@ func NewOutboundChannel(ctx context.Context, id ID, conn *grpc.ClientConn, opts // when a request is waiting. Otherwise, with eager reconnect, a draining stream // is replaced at once and a lost one after a backoff delay; without it, either // is replaced on the next request. +// +// A refused stream (see [isRefusal]) is replaced only after a backoff delay, +// with or without eager reconnect. Until the delay ends, each request fails +// at once with the refusal. Then, with eager reconnect, the channel opens the +// next stream; without it, the next request does. func (c *OutboundChannel) run() { defer c.queue.close() delay := eagerReconnectBaseDelay var retry <-chan time.Time var req *Request + var refusal error // the latest refusal; requests fail with it until refusedUntil + var refusedUntil time.Time // when the backoff after the latest refusal ends + suspect := false // no stream has proven viable since the latest refusal connect := true for { if !connect { @@ -120,6 +143,10 @@ func (c *OutboundChannel) run() { case <-c.ctx.Done(): return case r := <-c.queue.ch: + if refusal != nil && time.Now().Before(refusedUntil) { + c.queue.fail(r, refusal) + continue + } req = &r case <-retry: } @@ -130,7 +157,8 @@ func (c *OutboundChannel) run() { } return } - s, err := c.open() + refusal = nil + s, err := c.open(suspect) if err != nil { if c.ctx.Err() != nil { err = ErrNodeClosed @@ -141,16 +169,43 @@ func (c *OutboundChannel) run() { req.sendErrorResponse(c.id, err) req = nil } - retry, connect = c.pace(&delay), false + connect = false + if c.ctx.Err() == nil && isRefusal(err) { + // A failed check of the server, or a refusal of the stream. + refusal, suspect = err, true + retry, refusedUntil = c.backoff(&delay) + continue + } + retry = c.pace(&delay) continue } start := time.Now() req = s.sendLoop(req) c.setStreamUp(false) + if s.ended() { + // The session's stream is canceled, so receiving ends promptly + // and tells whether the server refused the stream. + <-s.verdict + refusal = s.refusal + } // A stream that received a frame or outlived the maximum delay has // proven viable, so the backoff starts over. if s.received.Load() || time.Since(start) >= eagerReconnectMaxDelay { delay = eagerReconnectBaseDelay + suspect = false + } + if refusal != nil { + // The calls pending on the refused stream have failed with the + // refusal. The request taken from the queue fails with it too, + // and so do later requests until the backoff ends. + c.recordHealth(refusal) + if req != nil { + c.queue.fail(*req, refusal) + req = nil + } + suspect, connect = true, false + retry, refusedUntil = c.backoff(&delay) + continue } connect = req != nil || s.draining.Load() && c.eagerReconnect retry = nil @@ -171,9 +226,24 @@ func (c *OutboundChannel) pace(delay *time.Duration) <-chan time.Time { return timer } +// backoff returns when the backoff after a refusal ends, with a timer for the +// next eager reconnect attempt at that time, and doubles delay up to the cap, +// with or without eager reconnect. Without eager reconnect the timer is nil, +// and the first request after the backoff opens the next stream. +func (c *OutboundChannel) backoff(delay *time.Duration) (<-chan time.Time, time.Time) { + until := time.Now().Add(*delay) + var timer <-chan time.Time + if c.eagerReconnect { + timer = time.After(*delay) + } + *delay = min(2*(*delay), eagerReconnectMaxDelay) + return timer, until +} + // open opens a stream, checks its remote end if the channel has a verifier, -// and starts the session's receive loop and GOAWAY watcher. -func (c *OutboundChannel) open() (*session, error) { +// and starts the session's receive loop and GOAWAY watcher. A suspect session +// follows a refused stream; see [session.viable]. +func (c *OutboundChannel) open(suspect bool) (*session, error) { ctx, cancel := context.WithCancel(c.ctx) st, err := NewGorumsClient(c.conn).NodeStream(ctx) if err != nil { @@ -193,10 +263,18 @@ func (c *OutboundChannel) open() (*session, error) { } } s := newSession(ctx, cancel, &c.endpoint, st, true, true) + s.suspect = suspect c.mu.Lock() c.sessions[s] = struct{}{} c.mu.Unlock() - c.setStreamUp(true) + if suspect { + // The stream counts as up only once it proves viable, so that a + // stream the server refuses again never counts as up. + s.onViable = func() { c.setStreamUp(true) } + s.probation = time.After(eagerReconnectMaxDelay) + } else { + c.setStreamUp(true) + } c.wg.Go(func() { _ = s.receive() c.mu.Lock() diff --git a/internal/stream/session.go b/internal/stream/session.go index b3b00bfc..17467070 100644 --- a/internal/stream/session.go +++ b/internal/stream/session.go @@ -3,6 +3,8 @@ package stream import ( "cmp" "context" + "errors" + "io" "sync/atomic" "time" ) @@ -39,8 +41,36 @@ type session struct { draining atomic.Bool sendStopped atomic.Bool received atomic.Bool // a frame has been received + + // verdict is closed when the stream delivers its first frame or receiving + // ends, whichever comes first. refusal is set before verdict is closed if + // the server refused an outbound session's stream; see [isRefusal]. + verdict chan struct{} + refusal error + // handoff decides which goroutine disposes of the pending calls when a + // stream ends before it delivered a frame; see [session.sendFailed]. + handoff atomic.Int32 + + // suspect is set on a session opened after a refused stream, before any + // stream proved viable; see [session.viable]. opened is when the session + // was created. + suspect bool + opened time.Time + // onViable, if non-nil, reports a suspect session's stream as up. The + // send loop calls it once the stream proves viable: the stream delivers + // a frame, or probation fires after [eagerReconnectMaxDelay]. + onViable func() + probation <-chan time.Time } +// Values of [session.handoff]. +const ( + streamFresh int32 = iota // no frame yet, and neither a send nor receiving failed + streamSendFailed // a send failed first; receiving disposes of the pending calls + streamAccepted // a frame arrived + streamClosed // receiving ended; receiving disposes of the pending calls +) + // newSession returns a session of e over stream that ends when ctx ends or cancel // is called. func newSession(ctx context.Context, cancel context.CancelFunc, e *endpoint, stream BidiStream, serverRequests, requeue bool) *session { @@ -52,15 +82,25 @@ func newSession(ctx context.Context, cancel context.CancelFunc, e *endpoint, str serverRequests: serverRequests, requeue: requeue, drain: make(chan struct{}), + verdict: make(chan struct{}), + opened: time.Now(), } } // sendLoop sends req, if non-nil, and then requests from the send queue, // until the session ends or starts draining. It returns the request it took // from the queue but did not send, or nil. +// +// On a suspect session with onViable, the send loop also reports the stream +// as up once it proves viable; see [session.checkViable]. func (s *session) sendLoop(req *Request) *Request { defer s.stopSending() + var verdict <-chan struct{} + if s.onViable != nil { + verdict = s.verdict + } for { + s.checkViable() if req == nil { select { case <-s.done: @@ -69,6 +109,12 @@ func (s *session) sendLoop(req *Request) *Request { return nil case r := <-s.queue.ch: req = &r + case <-verdict: + verdict = nil + continue + case <-s.probation: + s.markViable() + continue } } if s.ended() || s.draining.Load() { @@ -97,35 +143,142 @@ func (s *session) send(req Request) bool { err := s.stream.Send(req.Msg) s.sendStart.Store(0) if err != nil { - s.fail(err) + s.sendFailed(err) if !req.wantServerResponse() { req.sendErrorResponse(s.id, cmp.Or(req.Ctx.Err(), err)) } return false } - s.recordHealth(nil) + if s.viable() { + s.recordHealth(nil) + } if req.wantSendConfirmation() { req.deliver(response{NodeID: s.id}) } return true } +// viable reports whether a completed send shows that the session's stream +// works, so that it clears the channel's last error. A send on a stream that +// the server will refuse completes locally, before the refusal arrives. So, +// on a suspect session, only a stream that delivered a frame or has lived +// [eagerReconnectMaxDelay] proves viable. +func (s *session) viable() bool { + return !s.suspect || s.received.Load() || time.Since(s.opened) >= eagerReconnectMaxDelay +} + +// checkViable calls onViable, once, when the stream has delivered a frame or +// its probation has ended. +func (s *session) checkViable() { + if s.onViable == nil { + return + } + if s.received.Load() { + s.markViable() + return + } + select { + case <-s.probation: + s.markViable() + default: + } +} + +// markViable calls onViable unless the session has ended, and clears it. +func (s *session) markViable() { + if !s.ended() { + s.onViable() + } + s.onViable, s.probation = nil, nil +} + +// sendFailed ends the session after a send failed with err. +// +// On an outbound session whose stream has delivered no frame, a send that +// fails with io.EOF only cancels the stream: the server ended the stream, and +// only receiving learns its status, which tells whether the server refused +// the stream. The receive loop then disposes of the pending calls. It also +// does so if receiving has already ended. Otherwise, sendFailed ends the +// session as [session.fail] does. +func (s *session) sendFailed(err error) { + if s.requeue { + for { + state := s.handoff.Load() + if state == streamClosed { + s.cancel() + return + } + if state != streamFresh || !errors.Is(err, io.EOF) || s.ended() { + break + } + if s.handoff.CompareAndSwap(streamFresh, streamSendFailed) { + s.cancel() + return + } + } + } + s.fail(err) +} + // receive routes the frames received on the stream until receiving fails, then // ends the session and returns the receive error. func (s *session) receive() error { for { msg, err := s.stream.Recv() if err != nil { - s.fail(err) + s.receiveFailed(err) return err } s.recordHealth(nil) s.received.Store(true) + if s.handoff.Load() != streamAccepted { + s.accept() + } s.handle(msg) s.endIfDrained() } } +// accept records that the stream delivered its first frame, which shows that +// the server accepted the stream. If a send failed before the frame arrived, +// accept retries the pending calls, as for any lost stream. +func (s *session) accept() { + prev := s.handoff.Swap(streamAccepted) + close(s.verdict) + if prev == streamSendFailed { + s.end() + } +} + +// receiveFailed ends the session after receiving failed with err. On an +// outbound session whose stream delivered no frame, a refusal code in err +// (see [isRefusal]) means that the server refused the stream: the pending +// calls then fail with err instead of being retried. +func (s *session) receiveFailed(err error) { + prev := s.handoff.Swap(streamClosed) + if prev != streamAccepted { + if s.requeue && isRefusal(err) { + s.refusal = err + } + close(s.verdict) + } + switch { + case s.refusal != nil: + s.recordHealth(err) + s.cancel() + for _, req := range s.pending.drain() { + req.sendErrorResponse(s.id, err) + } + case prev == streamSendFailed: + // The failed send ended the session without recording its error; + // err is the stream's status. + s.recordHealth(err) + s.end() + default: + s.fail(err) + } +} + // handle routes one received frame. A frame in the peer's ID space is a new // request, queued for the handler; it is dropped if there is no handler. A // frame in this side's ID space is a response to a pending call; it is dropped @@ -193,6 +346,9 @@ func (s *session) ended() bool { // // An inbound channel has no next session, so on an inbound session, retry // fails every call with [ErrStreamDown]. +// +// The calls of a refused stream are not retried; see +// [session.receiveFailed]. func (s *session) retry(req Request) { switch { case s.requeue && !req.Streaming: diff --git a/internal/stream/session_test.go b/internal/stream/session_test.go index 13e2aed3..17515191 100644 --- a/internal/stream/session_test.go +++ b/internal/stream/session_test.go @@ -3,13 +3,16 @@ package stream import ( "context" "errors" + "io" "slices" "testing" "testing/synctest" "time" "github.com/relab/gorums/internal/testutils/mock" + "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" ) // newTestSession returns a session over a mock stream; it is not started. @@ -402,3 +405,226 @@ func TestSessionHandleBackChannel(t *testing.T) { s.handle(Message_builder{MessageSeqNo: 999, Method: mock.TestMethod}.Build()) }) } + +// scriptedStream is a [BidiStream] whose Send returns sendErr and whose Recv +// returns the frames and errors queued on recv, in order. +type scriptedStream struct { + sendErr error + recv chan scriptedRecv +} + +// scriptedRecv is one result of [scriptedStream.Recv]. +type scriptedRecv struct { + msg *Message + err error +} + +func (s *scriptedStream) Send(*Message) error { return s.sendErr } + +func (s *scriptedStream) Recv() (*Message, error) { + r := <-s.recv + return r.msg, r.err +} + +// frame queues msg for Recv. +func (s *scriptedStream) frame(msg *Message) { s.recv <- scriptedRecv{msg: msg} } + +// end queues err for Recv. +func (s *scriptedStream) end(err error) { s.recv <- scriptedRecv{err: err} } + +// newScriptedSession returns an outbound session, or an inbound one if +// requeue is false, over a stream whose Send returns sendErr; it is not +// started. +func newScriptedSession(t *testing.T, sendErr error, requeue bool) (*session, *scriptedStream) { + t.Helper() + e := newEndpoint(t.Context(), 1, 4, 0, nil, nil) + t.Cleanup(e.cancel) + ctx, cancel := context.WithCancel(e.ctx) + stream := &scriptedStream{sendErr: sendErr, recv: make(chan scriptedRecv, 4)} + return newSession(ctx, cancel, &e, stream, true, requeue), stream +} + +// responseErr returns the error of the response on ch, or fails the test if +// there is none. +func responseErr(t *testing.T, ch <-chan response, what string) error { + t.Helper() + select { + case resp := <-ch: + return resp.Err + default: + t.Fatalf("%s: no response", what) + return nil + } +} + +// TestSessionRefusalFailsPendingCalls verifies that an outbound session whose +// stream ends with a refusal code before it delivered a frame fails its +// pending calls with the refusal, and that every other end retries them. +func TestSessionRefusalFailsPendingCalls(t *testing.T) { + refusal := func(code codes.Code) error { return status.Error(code, "refused by test") } + tests := []struct { + name string + err error + afterFrame bool + inbound bool + wantCode codes.Code // the code both calls fail with; OK if retried + }{ + {name: "Unauthenticated", err: refusal(codes.Unauthenticated), wantCode: codes.Unauthenticated}, + {name: "PermissionDenied", err: refusal(codes.PermissionDenied), wantCode: codes.PermissionDenied}, + {name: "InvalidArgument", err: refusal(codes.InvalidArgument), wantCode: codes.InvalidArgument}, + {name: "Unimplemented", err: refusal(codes.Unimplemented), wantCode: codes.Unimplemented}, + {name: "Unavailable", err: refusal(codes.Unavailable)}, + {name: "Unknown", err: refusal(codes.Unknown)}, + {name: "Canceled", err: refusal(codes.Canceled)}, + {name: "EOF", err: io.EOF}, + {name: "RefusalAfterFrame", err: refusal(codes.Unauthenticated), afterFrame: true}, + {name: "Inbound", err: refusal(codes.Unauthenticated), inbound: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s, stream := newScriptedSession(t, nil, !tt.inbound) + twoWay := make(chan response, 1) + streaming := make(chan response, 1) + s.pending.add(1, Request{Ctx: t.Context(), Msg: &Message{}, ResponseChan: twoWay}) + s.pending.add(2, Request{Ctx: t.Context(), Msg: &Message{}, Streaming: true, ResponseChan: streaming}) + if tt.afterFrame { + stream.frame(Message_builder{MessageSeqNo: 999}.Build()) // an unknown response + } + stream.end(tt.err) + if err := s.receive(); !errors.Is(err, tt.err) { + t.Fatalf("receive() = %v, want %v", err, tt.err) + } + + calls := map[string]chan response{"two-way call": twoWay, "streaming call": streaming} + switch { + case tt.wantCode != codes.OK: + for what, ch := range calls { + if got := status.Code(responseErr(t, ch, what)); got != tt.wantCode { + t.Errorf("%s failed with code %v, want %v", what, got, tt.wantCode) + } + } + if got := len(s.queue.ch); got != 0 { + t.Errorf("requeued %d calls, want 0", got) + } + case tt.inbound: + for what, ch := range calls { + if err := responseErr(t, ch, what); !errors.Is(err, ErrStreamDown) { + t.Errorf("%s failed with %v, want %v", what, err, ErrStreamDown) + } + } + default: + if got := len(s.queue.ch); got != 1 { + t.Errorf("requeued %d calls, want the two-way call", got) + } + if err := responseErr(t, streaming, "streaming call"); !errors.Is(err, ErrStreamDown) { + t.Errorf("streaming call failed with %v, want %v", err, ErrStreamDown) + } + } + }) + } +} + +// TestSessionFailedSendLeavesCallsToReceive verifies that a send that fails +// with io.EOF before the stream delivered a frame leaves the pending calls to +// the receive loop, which learns the stream's status: a refusal fails them, +// and any other end, or a frame that shows that the server accepted the +// stream, retries them. +func TestSessionFailedSendLeavesCallsToReceive(t *testing.T) { + tests := []struct { + name string + recv func(*scriptedStream) + wantCode codes.Code // OK if retried + wantRequeued bool + }{ + { + name: "Refused", + recv: func(st *scriptedStream) { st.end(status.Error(codes.Unauthenticated, "refused")) }, + wantCode: codes.Unauthenticated, + }, + { + name: "Lost", + recv: func(st *scriptedStream) { st.end(status.Error(codes.Unavailable, "lost")) }, + wantRequeued: true, + }, + { + name: "FrameAfterFailedSend", + recv: func(st *scriptedStream) { + st.frame(Message_builder{MessageSeqNo: 1}.Build()) // the call's response + st.end(status.Error(codes.Unauthenticated, "refused after a frame")) + }, + wantRequeued: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s, stream := newScriptedSession(t, io.EOF, true) + responseChan := make(chan response, 1) + s.queue.push(Request{Ctx: t.Context(), Msg: Message_builder{MessageSeqNo: 1}.Build(), ResponseChan: responseChan}, false) + if req := s.sendLoop(nil); req != nil { + t.Fatalf("send loop returned request %d, want none", req.Msg.GetMessageSeqNo()) + } + if !s.ended() { + t.Fatal("session did not end after its send failed") + } + if len(responseChan) != 0 || len(s.queue.ch) != 0 { + t.Fatal("the failed send disposed of the pending call; want it left to the receive loop") + } + + tt.recv(stream) + _ = s.receive() + if got := len(s.queue.ch) == 1; got != tt.wantRequeued { + t.Errorf("requeued = %t, want %t", got, tt.wantRequeued) + } + if tt.wantCode != codes.OK { + if got := status.Code(responseErr(t, responseChan, "call")); got != tt.wantCode { + t.Errorf("call failed with code %v, want %v", got, tt.wantCode) + } + } else if got := len(responseChan); got != 0 { + t.Errorf("call got %d responses, want it retried", got) + } + }) + } +} + +// TestSessionVerdict verifies that a session's verdict is closed once the +// stream delivers its first frame or receiving ends, and that it reports a +// refusal only for a refused stream. +func TestSessionVerdict(t *testing.T) { + tests := []struct { + name string + recv func(*scriptedStream) + wantRefusal codes.Code // OK for none + }{ + {name: "Frame", recv: func(st *scriptedStream) { st.frame(Message_builder{MessageSeqNo: 999}.Build()) }}, + {name: "Refused", recv: func(st *scriptedStream) { st.end(status.Error(codes.PermissionDenied, "refused")) }, wantRefusal: codes.PermissionDenied}, + {name: "Lost", recv: func(st *scriptedStream) { st.end(status.Error(codes.Unavailable, "lost")) }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, stream := newScriptedSession(t, nil, true) + go func() { _ = s.receive() }() + defer func() { + stream.end(context.Canceled) // ends a receive loop still running + synctest.Wait() + }() + synctest.Wait() + select { + case <-s.verdict: + t.Fatal("verdict closed before the stream delivered anything") + default: + } + tt.recv(stream) + synctest.Wait() + select { + case <-s.verdict: + default: + t.Fatal("verdict not closed") + } + if got := status.Code(s.refusal); got != tt.wantRefusal { + t.Errorf("refusal = %v, want code %v", s.refusal, tt.wantRefusal) + } + }) + }) + } +} diff --git a/internal/tests/tls/peer_identity_test.go b/internal/tests/tls/peer_identity_test.go index a6dbfe0f..01d43ecc 100644 --- a/internal/tests/tls/peer_identity_test.go +++ b/internal/tests/tls/peer_identity_test.go @@ -1,8 +1,11 @@ package tls import ( + "context" "crypto/tls" + "errors" "fmt" + "slices" "strings" "sync/atomic" "testing" @@ -142,7 +145,9 @@ func TestPeerIdentityWrongCertificate(t *testing.T) { if want := "gorums: node 2 failed verification"; !strings.Contains(err.Error(), want) { t.Errorf("call from server 1 to node 2: err = %q; want it to contain %q", err, want) } - // Inbound check: server 2 dials server 3 and claims ID 2 with node 3's certificate. + // Inbound check: server 2 dials server 3 and claims ID 2 with + // node 3's certificate; its calls fail with server 3's refusal. + awaitCode(t, servers[1].PeerConfig().Node(3), codes.Unauthenticated) awaitLastErrCode(t, servers[1].PeerConfig().Node(3), codes.Unauthenticated) // The other pair is unaffected. if got, err := remoteCall(t, servers[0].PeerConfig().Node(3), 3*time.Second); err != nil || got != "server 3 from 1: hello" { @@ -152,6 +157,81 @@ func TestPeerIdentityWrongCertificate(t *testing.T) { } } +// TestPeerIdentityRefusalIsPaced verifies that a refused peer's calls fail +// with the refusal at once, and that the refused peer paces its new streams, +// so the refusing server runs its verifier only a few times. +func TestPeerIdentityRefusalIsPaced(t *testing.T) { + for _, mode := range peerIdentityModes { + t.Run(mode.name, func(t *testing.T) { + var checks atomic.Int32 + verify := peerVerifier(3) + countOnServer3 := gorums.WithLocalServerOptionsFunc(func(id gorums.ID) []gorums.ServerOption { + if id != 3 { + return nil + } + return []gorums.ServerOption{gorums.WithPeerVerifier(func(ctx context.Context, id gorums.ID) error { + checks.Add(1) + return verify(ctx, id) + })} + }) + // Server 2 presents node 3's certificate, so server 3 refuses + // server 2's claim. + servers, _ := peerGroup(t, newTestCA(t), 3, []gorums.ID{1, 3, 3}, append(mode.opts, countOnServer3)...) + node3 := servers[1].PeerConfig().Node(3) + awaitCode(t, node3, codes.Unauthenticated) + + before := checks.Load() + const window = time.Second + calls := 0 + for deadline := time.Now().Add(window); time.Now().Before(deadline); calls++ { + if _, err := remoteCall(t, node3, 3*time.Second); status.Code(err) != codes.Unauthenticated { + t.Fatalf("call %d from server 2 to node 3: err = %v; want code %v", calls+1, err, codes.Unauthenticated) + } + } + t.Logf("%d refused calls, %d verifier runs on server 3", calls, checks.Load()-before) + if got := checks.Load() - before; got > 20 { + t.Errorf("server 3 ran its verifier %d times during %d refused calls in %v; want at most 20", got, calls, window) + } + }) + } +} + +// TestPeerIdentityRefusedPeerNotConnected verifies that, under stream +// deduplication, a server whose stream to one peer is refused does not count +// that peer as connected, so its WaitForAll does not return. +func TestPeerIdentityRefusedPeerNotConnected(t *testing.T) { + verify := peerVerifier(3) + // Server 3 refuses server 2's claim only. Its check takes a while, as a + // real check may, so each refused stream would count as up meanwhile. + refuse2On3 := gorums.WithLocalServerOptionsFunc(func(id gorums.ID) []gorums.ServerOption { + if id != 3 { + return nil + } + return []gorums.ServerOption{gorums.WithPeerVerifier(func(ctx context.Context, id gorums.ID) error { + if id == 2 { + time.Sleep(20 * time.Millisecond) + return errors.New("server 2 is not trusted") + } + return verify(ctx, id) + })} + }) + servers, _ := peerGroup(t, newTestCA(t), 3, nil, gorums.WithStreamDedup(), refuse2On3) + server2 := servers[1] + awaitCode(t, server2.PeerConfig().Node(3), codes.Unauthenticated) + // The first stream counted as up until its refusal arrived. + if !gorumstest.WaitUntil(t, 3*time.Second, func() bool { + return !slices.Contains(server2.ConnectedPeers().NodeIDs(), 3) + }) { + t.Fatalf("server 2 ConnectedPeers = %v; want it without node 3", server2.ConnectedPeers().NodeIDs()) + } + + // Later streams, which server 3 refuses too, never count as up. + ctx := gorumstest.Context(t, 500*time.Millisecond) + if _, err := server2.WaitForAll(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("server 2 WaitForAll = %v; want %v, since server 3 refuses server 2", err, context.DeadlineExceeded) + } +} + // TestPeerIdentityClients verifies how a server with peer identity checks // treats clients: a client that claims a peer's ID without that peer's // certificate is refused and cannot take over the peer's stream, and a @@ -200,11 +280,10 @@ func TestPeerIdentityClients(t *testing.T) { } return } - // The client channel resends the call on each new stream until - // the call's context ends, so the call fails by its deadline; the - // refusal's status is the node's last error. - if _, err := remoteCall(t, node, time.Second); err == nil { - t.Fatal("call from a refused client succeeded; want failure") + // The call fails with the server's refusal, which is also the + // node's last error. + if _, err := remoteCall(t, node, 3*time.Second); status.Code(err) != codes.Unauthenticated { + t.Fatalf("call from a refused client: err = %v; want code %v", err, codes.Unauthenticated) } awaitLastErrCode(t, node, codes.Unauthenticated) if n := calls[1].Load() - before; n != 0 { diff --git a/server_internal_test.go b/server_internal_test.go index 81d2da96..a6252861 100644 --- a/server_internal_test.go +++ b/server_internal_test.go @@ -16,8 +16,10 @@ import ( "github.com/relab/gorums/internal/testutils/servers" "go.uber.org/goleak" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" pb "google.golang.org/protobuf/types/known/wrapperspb" ) @@ -263,14 +265,14 @@ func TestSelfNodeIDStreamRejectedEndToEnd(t *testing.T) { return srv }) - // The client claims the server's own node ID (1). Its stream is rejected, - // so the call cannot complete. + // The client claims the server's own node ID (1). Its stream is rejected + // with codes.InvalidArgument, and the call fails with that refusal. cfg, _ := connectAsPeer(t, 1, addrs) node := cfg.Nodes()[0] - // Bound the request itself so it fails fast; wait for that failure on a - // separate, longer deadline so the two do not race in the select. - reqCtx := testTimeoutContext(t, 500*time.Millisecond) + // Bound the request by more than the wait below, so that the call fails + // with the refusal, not with its deadline. + reqCtx := testTimeoutContext(t, 5*time.Second) reqMsg, err := stream.NewMessage(reqCtx, conn.NodeTransport(node).NextMsgID(), mock.TestMethod, pb.String("hello")) if err != nil { t.Fatalf("NewMessage() error: %v", err) @@ -280,8 +282,8 @@ func TestSelfNodeIDStreamRejectedEndToEnd(t *testing.T) { select { case resp := <-responseChan: - if resp.Err == nil { - t.Fatal("call over a self-ID stream succeeded; want failure") + if got := status.Code(resp.Err); got != codes.InvalidArgument { + t.Fatalf("call over a self-ID stream: err = %v; want code %v", resp.Err, codes.InvalidArgument) } case <-time.After(3 * time.Second): t.Fatal("call over a self-ID stream neither failed nor completed") diff --git a/server_options.go b/server_options.go index 41b38650..db612975 100644 --- a/server_options.go +++ b/server_options.go @@ -146,19 +146,22 @@ func WithStreamDedup() ServerOption { // - A stream that the server opens to a peer is checked against the dialed // node's ID before any request is sent on it. If the check fails, the // stream is closed, the waiting call fails with codes.Unauthenticated, -// and the server tries again later, as for a failed connection. +// and the server backs off before it opens a new stream to that peer, as +// a refused peer does. // // When the server refuses a stream that a peer opened, the refusal's status // carries the text of verify's error, and the refused peer sees it in -// [Node.LastErr]. The refused peer resends its pending two-way calls on new -// streams until their contexts end, so calls between peers need deadlines; its -// streaming calls fail instead. When a -// stream that the server opened fails the check, the waiting call fails with -// the check's error. The server runs verify once per stream, outside its -// internal locks. With WithPeerVerifier, [ServerContext.SenderID] is -// authenticated for every configured peer. [NewServer] panics if verify is nil -// or if the server has no [WithPeers]. Use [TLSPeerVerifier] to bind node IDs -// to mTLS certificates. +// [Node.LastErr] until one of its streams to the server succeeds. The refused +// peer's calls on that stream fail with this status; they are not resent. The +// refused peer then backs off before it opens a new stream to the server: +// 50 ms after the first refusal, doubling after each further refusal up to +// 2 s. Until the backoff ends, its calls to the server fail at once with the +// refusal. When a stream that the server opened fails the check, the waiting +// call fails with the check's error. The server runs verify once per stream, +// outside its internal locks. With WithPeerVerifier, [ServerContext.SenderID] +// is authenticated for every configured peer. [NewServer] panics if verify is +// nil or if the server has no [WithPeers]. Use [TLSPeerVerifier] to bind node +// IDs to mTLS certificates. func WithPeerVerifier(verify PeerVerifier) ServerOption { return func(o *serverOptions) { o.peerVerifier = verify diff --git a/server_test.go b/server_test.go index 11be2478..b1873ad8 100644 --- a/server_test.go +++ b/server_test.go @@ -1455,11 +1455,12 @@ func TestServerPeerVerifierRefusals(t *testing.T) { t.Run("UnknownPeerIDRefused", func(t *testing.T) { servers := start(t, acceptAll) node := client(t, servers[0], gorums.WithMetadata(metadata.Pairs("gorums-node-id", "99"))) - if err := call(t, node, 500*time.Millisecond); err == nil { - t.Fatal("call that claims an unknown node ID succeeded; want failure") - } // The server refuses the stream, which ends with its status; the - // client's node records it as its last error. + // call fails with it, and the client's node records it as its last + // error. + if err := call(t, node, 3*time.Second); status.Code(err) != codes.Unauthenticated { + t.Fatalf("call that claims an unknown node ID: err = %v; want code %v", err, codes.Unauthenticated) + } if !gorumstest.WaitUntil(t, 3*time.Second, func() bool { return status.Code(node.LastErr()) == codes.Unauthenticated }) {