From 7e88636a7afd93aa9f47ca30c30a44b6de82985c Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Tue, 11 Aug 2026 18:45:07 +0200 Subject: [PATCH 1/5] gorums: replace call options with typed call handles Every generated call took a variadic opts ...CallOption, which held two unrelated things. IgnoreErrors switched a one-way call between blocking and fire-and-forget, and Interceptors carried type-erased interceptors that CallContext had to assert back to their concrete type at dispatch. Neither is expressible in the signature, so a caller could pass an interceptor for the wrong message types and find out at run time. A quorum call now returns *Call[Req, Resp], which embeds *Responses[Resp] so the terminal methods are unchanged, and adds Intercept for interceptors typed to the call's own request and response. A one-way call returns *OnewayCall[Req] with the two behaviors as named terminals: Send blocks until every send completes and reports the failures, Async dispatches and defers them to Wait. Async also does what IgnoreErrors could not: keep several one-way calls in flight from a single goroutine and still learn whether they were sent. Dropping a one-way handle without consuming it sends nothing, so a call is now dispatched only where the code says so. Consuming a handle twice panics rather than re-sending the request or blocking on confirmations the first dispatch already drained. Intercept after dispatch panics too, since an interceptor can no longer affect an in-flight call; async and correctable calls mark dispatch before starting their goroutine so that panic is deterministic rather than a race. ErrSkipNode no longer counts as a node error in Threshold or Correctable. A node the caller's own request transform skipped is neither a success nor a failure, and counting it as an error made a deliberate skip look like an outage. The type parameters are constrained by proto.Message directly, replacing the msg alias that hid what the constraint was, and newResponses is unexported now that Call constructs it. --- call.go | 212 +++++++++++ call_async_test.go | 39 +- call_client_interceptor_test.go | 60 ++- call_context.go | 156 +++----- call_quorum_test.go | 16 +- call_test.go | 347 ++++++++++++++++++ callopts.go | 50 --- callopts_test.go | 135 ------- .../gengorums/template_multicast.go | 16 +- .../gengorums/template_quorumcall.go | 10 +- .../gengorums/template_unicast.go | 17 +- correctable.go | 10 +- examples/storage/client.go | 4 +- examples/storage/repl.go | 4 +- examples/storage/server.go | 2 +- inbound_manager_test.go | 19 +- internal/tests/oneway/oneway_test.go | 269 +++++++------- multicast.go | 54 +-- quorumcall.go | 51 +-- remote_call_test.go | 96 ----- remote_call.go => remotecall.go | 7 +- responses.go | 32 +- responses_test.go | 61 +-- server_e2e_test.go | 16 +- unicast.go | 48 +-- 25 files changed, 973 insertions(+), 758 deletions(-) create mode 100644 call.go create mode 100644 call_test.go delete mode 100644 callopts.go delete mode 100644 callopts_test.go delete mode 100644 remote_call_test.go rename remote_call.go => remotecall.go (79%) diff --git a/call.go b/call.go new file mode 100644 index 000000000..51004c89b --- /dev/null +++ b/call.go @@ -0,0 +1,212 @@ +package gorums + +import ( + "errors" + "sync" + + "github.com/relab/gorums/internal/stream" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/emptypb" +) + +// Call represents a lazily dispatched quorum call. +// Register interceptors with [Call.Intercept] before consuming its responses. +// A Call may be consumed once. +type Call[Req, Resp proto.Message] struct { + *Responses[Resp] + ctx *CallContext[Req, Resp] +} + +// Intercept registers interceptors for this call, applied in call-site order. +// It returns the same handle for fluent chaining: +// +// resp, err := storage.ReadQC(ctx, req).Intercept(logging, audit).Majority() +// +// Intercept must be called before any consuming method (a terminal method, +// async terminal, correctable call, or ranging Results). Calling it after +// dispatch has started panics. Nil interceptors are ignored. +func (c *Call[Req, Resp]) Intercept(ics ...ClientInterceptor[Req, Resp]) *Call[Req, Resp] { + c.ctx.intercept(ics...) + // The interceptors may have wrapped the response sequence; re-sync the + // embedded Responses so its terminal methods observe the wrapped sequence. + c.Responses.seq = c.ctx.responseSeq + return c +} + +// ClientInterceptor intercepts and processes quorum calls, allowing modification of +// requests, responses, and aggregation logic. Interceptors can be chained together. +// +// Type parameters: +// - Req: The request message type sent to nodes +// - Resp: The response message type from individual nodes +// +// The interceptor receives the CallContext for metadata access, the current response +// iterator (next), and returns a new response iterator. This pattern allows +// interceptors to wrap the response stream with custom logic. +// +// Custom interceptors can be created like this: +// +// func LoggingInterceptor[Req, Resp proto.Message]( +// ctx *gorums.CallContext[Req, Resp], +// next gorums.ResponseSeq[Resp], +// ) gorums.ResponseSeq[Resp] { +// return func(yield func(gorums.NodeResponse[Resp]) bool) { +// for resp := range next { +// log.Printf("Response from node %d", resp.NodeID) +// if !yield(resp) { return } +// } +// } +// } +type ClientInterceptor[Req, Resp proto.Message] func(ctx *CallContext[Req, Resp], next ResponseSeq[Resp]) ResponseSeq[Resp] + +// MapRequest returns an interceptor that applies per-node request transformations. +// Multiple interceptors can be chained together, with transforms applied in order. +// +// The fn receives the original request and a node, and returns the transformed +// request to send to that node. If the function returns an invalid message or nil, +// an ErrSkipNode error is sent for that node, indicating it was skipped. +func MapRequest[Req, Resp proto.Message](fn func(Req, *Node) Req) ClientInterceptor[Req, Resp] { + return func(ctx *CallContext[Req, Resp], next ResponseSeq[Resp]) ResponseSeq[Resp] { + if fn != nil { + ctx.reqTransforms = append(ctx.reqTransforms, fn) + } + return next + } +} + +// MapResponse returns an interceptor that applies per-node response transformations. +// +// The fn receives the response from a node and the node itself, and returns the +// transformed response. +func MapResponse[Req, Resp proto.Message](fn func(Resp, *Node) Resp) ClientInterceptor[Req, Resp] { + return func(ctx *CallContext[Req, Resp], next ResponseSeq[Resp]) ResponseSeq[Resp] { + if fn == nil { + return next + } + // Wrap the response iterator with the transformation logic. + return func(yield func(NodeResponse[Resp]) bool) { + for resp := range next { + // We only apply the transformation if there is no error. + // Errors are passed through as-is. + if resp.Err == nil { + if node := ctx.Node(resp.NodeID); node != nil { + resp.Value = fn(resp.Value, node) + } + } + if !yield(resp) { + return + } + } + } + } +} + +// OnewayCall represents a lazily dispatched multicast or unicast call. +// [OnewayCall.Send] and [OnewayCall.Async] each consume the call; invoking +// either after the call has been consumed panics. A handler that dispatches a +// one-way call back to its callers should call [ServerContext.Release] first. +type OnewayCall[Req proto.Message] struct { + ctx *CallContext[Req, *emptypb.Empty] + unicast bool +} + +// Intercept registers interceptors for this one-way call, applied in call-site +// order, and returns the same handle for fluent chaining. It must be called +// before [OnewayCall.Send] or [OnewayCall.Async]; calling it after dispatch +// panics. Nil interceptors are ignored. Only request transforms (see +// [MapRequest]) take effect for one-way calls, since no responses are collected. +func (c *OnewayCall[Req]) Intercept(ics ...ClientInterceptor[Req, *emptypb.Empty]) *OnewayCall[Req] { + c.ctx.intercept(ics...) + return c +} + +// Send dispatches the request and blocks until every message has reached its +// node's stream. A one-way call carries no reply, so there is nothing further +// to await. +// +// For multicast, Send returns nil only if the send completes for every target +// node; send failures are returned as a [QuorumCallError] with cause +// [ErrSendFailure] and per-node errors. For unicast, Send returns the single +// send error, or the context error if the context is cancelled first. +// +// A server handler dispatching a back-channel call should release its hold on +// the server first, so that inbound processing is not blocked while the send +// completes. Use [OnewayCall.Async] to keep several sends in flight from a +// single goroutine. +// +// Send consumes the call; calling it again on the same handle panics. +func (c *OnewayCall[Req]) Send() error { + c.dispatch() + return c.collect() +} + +// Async dispatches the request without waiting for the sends to complete, so a +// single goroutine can keep several one-way calls in flight: +// +// h := Multicast(ctx, msg).Async() +// // ... dispatch more calls ... +// err := h.Wait() +// +// Async starts no goroutine; [OnewayAsync.Wait] collects the send confirmations +// on the caller's goroutine. Dropping the handle without calling Wait is safe. +// +// Async consumes the call; calling it again on the same handle panics. +func (c *OnewayCall[Req]) Async() *OnewayAsync { + c.dispatch() + return &OnewayAsync{collect: c.collect} +} + +// dispatch installs the reply channel and sends the request exactly once. +// It panics if the handle was already consumed. +func (c *OnewayCall[Req]) dispatch() { + if c.ctx.dispatched.Swap(true) { + panic("gorums: OnewayCall.Send or OnewayCall.Async called more than once on the same handle") + } + c.ctx.replyChan = make(chan NodeResponse[*stream.Message], c.ctx.config.Size()) + c.ctx.sendOnce.Do(c.ctx.send) +} + +// collect gathers one send confirmation per node and reports the failures, +// aggregating them for multicast and passing the single error through for +// unicast. Nodes skipped by a request transform are not failures. +func (c *OnewayCall[Req]) collect() error { + if c.unicast { + select { + case r := <-c.ctx.replyChan: + return r.Err + case <-c.ctx.Done(): + return c.ctx.Err() + } + } + var errs []nodeError + for range c.ctx.config.Size() { + select { + case r := <-c.ctx.replyChan: + if r.Err != nil && !errors.Is(r.Err, ErrSkipNode) { + errs = append(errs, nodeError{cause: r.Err, nodeID: r.NodeID}) + } + case <-c.ctx.Done(): + return c.ctx.Err() + } + } + if len(errs) > 0 { + return QuorumCallError{cause: ErrSendFailure, errors: errs} + } + return nil +} + +// OnewayAsync is the send-completion handle of a one-way call dispatched with +// [OnewayCall.Async]. +type OnewayAsync struct { + collect func() error + once sync.Once + err error +} + +// Wait blocks until send completion is known and returns the same error +// [OnewayCall.Send] would have returned. It may be called more than once and +// returns the same result each time. +func (a *OnewayAsync) Wait() error { + a.once.Do(func() { a.err = a.collect() }) + return a.err +} diff --git a/call_async_test.go b/call_async_test.go index 1ae836994..01a28e1c0 100644 --- a/call_async_test.go +++ b/call_async_test.go @@ -52,7 +52,7 @@ func TestAsync(t *testing.T) { mock.TestMethod, ) - future := tt.call(responses) + future := tt.call(responses.Responses) reply, err := future.Get() if !checkQuorumCall(t, err, nil) { @@ -85,6 +85,43 @@ func TestAsync_Error(t *testing.T) { } } +// TestAsyncDone verifies that Async.Done reports false while the call is still +// in flight and true once a result is available. +func TestAsyncDone(t *testing.T) { + t.Run("PendingReportsNotDone", func(t *testing.T) { + // A call over a never-dialed config never completes, so the future + // stays pending and Done reports false. + config := gorumstest.NoDialedConfig(t) + ctx := gorumstest.Context(t, 2*time.Second) + future := gorums.QuorumCall[*pb.StringValue, *pb.StringValue]( + config.Context(ctx), + pb.String("test"), + mock.TestMethod, + ).AsyncMajority() + + if future.Done() { + t.Error("Done() = true for a call that has not completed, want false") + } + }) + + t.Run("CompletedReportsDone", func(t *testing.T) { + config := gorumstest.Config(t, 3, gorumstest.EchoServerFn) + ctx := gorumstest.Context(t, 2*time.Second) + future := gorums.QuorumCall[*pb.StringValue, *pb.StringValue]( + config.Context(ctx), + pb.String("test"), + mock.TestMethod, + ).AsyncMajority() + + if _, err := future.Get(); err != nil { + t.Fatalf("Get() error: %v", err) + } + if !future.Done() { + t.Error("Done() = false after Get() returned, want true") + } + }) +} + func BenchmarkAsyncQuorumCall(b *testing.B) { for _, numNodes := range []int{3, 5, 7, 9} { config := gorumstest.Config(b, numNodes, gorumstest.EchoServerFn) diff --git a/call_client_interceptor_test.go b/call_client_interceptor_test.go index a9e47d430..42790e674 100644 --- a/call_client_interceptor_test.go +++ b/call_client_interceptor_test.go @@ -73,8 +73,7 @@ func TestCustomLoggingInterceptor(t *testing.T) { config.Context(ctx), pb.String("test"), mock.TestMethod, - gorums.Interceptors(LoggingInterceptor[*pb.StringValue, *pb.StringValue]), - ) + ).Intercept(LoggingInterceptor[*pb.StringValue, *pb.StringValue]) result, err := responses.Majority() if err != nil { @@ -96,12 +95,11 @@ func TestCustomFilterInterceptor(t *testing.T) { config.Context(ctx), pb.String("test"), mock.TestMethod, - gorums.Interceptors(FilterInterceptor[*pb.StringValue]( - func(resp gorums.NodeResponse[*pb.StringValue]) bool { - return resp.Err == nil // Only keep successful responses - }, - )), - ) + ).Intercept(FilterInterceptor[*pb.StringValue]( + func(resp gorums.NodeResponse[*pb.StringValue]) bool { + return resp.Err == nil // Only keep successful responses + }, + )) result, err := responses.First() if err != nil { @@ -124,10 +122,9 @@ func TestInterceptorChaining(t *testing.T) { config.Context(ctx), pb.String("test"), mock.TestMethod, - gorums.Interceptors( - LoggingInterceptor[*pb.StringValue, *pb.StringValue], - CountingInterceptor[*pb.StringValue, *pb.StringValue](&count), - ), + ).Intercept( + LoggingInterceptor[*pb.StringValue, *pb.StringValue], + CountingInterceptor[*pb.StringValue, *pb.StringValue](&count), ) result, err := responses.Majority() @@ -156,15 +153,14 @@ func TestCustomInterceptorWithMapRequest(t *testing.T) { config.Context(ctx), pb.String("test"), mock.TestMethod, - gorums.Interceptors( - // Custom: count responses - CountingInterceptor[*pb.StringValue, *pb.StringValue](&count), - // Built-in: transform request (identity transform for this test) - gorums.MapRequest[*pb.StringValue, *pb.StringValue]( - func(req *pb.StringValue, _ *gorums.Node) *pb.StringValue { - return req - }, - ), + ).Intercept( + // Custom: count responses + CountingInterceptor[*pb.StringValue, *pb.StringValue](&count), + // Built-in: transform request (identity transform for this test) + gorums.MapRequest[*pb.StringValue, *pb.StringValue]( + func(req *pb.StringValue, _ *gorums.Node) *pb.StringValue { + return req + }, ), ) @@ -214,12 +210,11 @@ func BenchmarkQuorumCallMapRequest(b *testing.B) { cfgCtx, pb.String("benchmark payload"), mock.TestMethod, - gorums.Interceptors( - gorums.MapRequest[*pb.StringValue, *pb.StringValue]( - func(req *pb.StringValue, _ *gorums.Node) *pb.StringValue { - return req - }, - ), + ).Intercept( + gorums.MapRequest[*pb.StringValue, *pb.StringValue]( + func(req *pb.StringValue, _ *gorums.Node) *pb.StringValue { + return req + }, ), ) if _, err := responses.Majority(); err != nil { @@ -236,12 +231,11 @@ func BenchmarkQuorumCallMapRequest(b *testing.B) { cfgCtx, pb.String("benchmark payload"), mock.TestMethod, - gorums.Interceptors( - gorums.MapRequest[*pb.StringValue, *pb.StringValue]( - func(req *pb.StringValue, n *gorums.Node) *pb.StringValue { - return pb.String(fmt.Sprintf("%s-node-%d", req.GetValue(), n.ID())) - }, - ), + ).Intercept( + gorums.MapRequest[*pb.StringValue, *pb.StringValue]( + func(req *pb.StringValue, n *gorums.Node) *pb.StringValue { + return pb.String(fmt.Sprintf("%s-node-%d", req.GetValue(), n.ID())) + }, ), ) if _, err := responses.Majority(); err != nil { diff --git a/call_context.go b/call_context.go index b20e63119..cacbdf21c 100644 --- a/call_context.go +++ b/call_context.go @@ -4,41 +4,16 @@ import ( "context" "slices" "sync" + "sync/atomic" "github.com/relab/gorums/internal/stream" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/emptypb" ) -// ClientInterceptor intercepts and processes quorum calls, allowing modification of -// requests, responses, and aggregation logic. Interceptors can be chained together. -// -// Type parameters: -// - Req: The request message type sent to nodes -// - Resp: The response message type from individual nodes -// -// The interceptor receives the CallContext for metadata access, the current response -// iterator (next), and returns a new response iterator. This pattern allows -// interceptors to wrap the response stream with custom logic. -// -// Custom interceptors can be created like this: -// -// func LoggingInterceptor[Req, Resp proto.Message]( -// ctx *gorums.CallContext[Req, Resp], -// next gorums.ResponseSeq[Resp], -// ) gorums.ResponseSeq[Resp] { -// return func(yield func(gorums.NodeResponse[Resp]) bool) { -// for resp := range next { -// log.Printf("Response from node %d", resp.NodeID) -// if !yield(resp) { return } -// } -// } -// } -type ClientInterceptor[Req, Resp msg] func(ctx *CallContext[Req, Resp], next ResponseSeq[Resp]) ResponseSeq[Resp] - // CallContext provides context and access to the quorum call state for interceptors. // It exposes the request, configuration, metadata about the call, and the response iterator. -type CallContext[Req, Resp msg] struct { +type CallContext[Req, Resp proto.Message] struct { context.Context config Config request Req @@ -63,21 +38,51 @@ type CallContext[Req, Resp msg] struct { // call to Responses(). This deferred sending allows interceptors // to register request transformations before dispatch. sendOnce sync.Once + + // dispatched is set once dispatch has been initiated (by sendNow or by + // marking an async/correctable call). Once set, Intercept panics because + // interceptors can no longer affect the in-flight call. It is an + // atomic.Bool rather than a plain bool because sendNow can be called again, + // redundantly, from the goroutine an async or correctable call spawns + // (ranging over responseSeq calls sendNow), concurrently with a caller + // checking or setting the flag on another goroutine. + dispatched atomic.Bool } // sendNow triggers request dispatch exactly once. func (c *CallContext[Req, Resp]) sendNow() { + c.markDispatched() c.sendOnce.Do(c.send) } +// markDispatched records that dispatch has been initiated, so a later Intercept +// panics. It is idempotent and does not itself send anything. +func (c *CallContext[Req, Resp]) markDispatched() { + c.dispatched.Store(true) +} + +// intercept applies the given interceptors in order, before dispatch. Nil +// interceptors are ignored. It panics if the call has already been dispatched, +// since interceptors can no longer influence an in-flight call. +func (c *CallContext[Req, Resp]) intercept(ics ...ClientInterceptor[Req, Resp]) { + if c.dispatched.Load() { + panic("gorums: Intercept called after the call was dispatched") + } + for _, ic := range ics { + if ic == nil { + continue + } + c.responseSeq = ic(c, c.responseSeq) + } +} + // newQuorumCallContext constructs a CallContext for quorum calls (two-way, always returns responses). // A reply channel is always created; streaming controls both its buffer size and the response iterator type. -func newQuorumCallContext[Req, Resp msg]( +func newQuorumCallContext[Req, Resp proto.Message]( ctx *ConfigContext, req Req, method string, streaming bool, - interceptors []any, ) *CallContext[Req, Resp] { config := ctx.Config() n := config.Size() @@ -98,37 +103,28 @@ func newQuorumCallContext[Req, Resp msg]( } else { clientCtx.responseSeq = clientCtx.defaultResponseSeq() } - clientCtx.applyInterceptors(interceptors) return clientCtx } -// newMulticastCallContext constructs a CallContext for multicast (one-way, no responses). -// A reply channel is created only when waitForSend=true (blocking send); fire-and-forget -// calls receive a nil channel, meaning no router entry is registered. -func newMulticastCallContext[Req msg]( - ctx *ConfigContext, +// newOnewayCallContext constructs a CallContext for a one-way call over config. +// The reply channel is installed by [OnewayCall.dispatch], since only a consumed +// handle collects send confirmations. +func newOnewayCallContext[Req proto.Message]( + ctx context.Context, + config Config, req Req, method string, - waitForSend bool, - interceptors []any, ) *CallContext[Req, *emptypb.Empty] { - config := ctx.Config() - var replyChan chan NodeResponse[*stream.Message] - if waitForSend { - replyChan = make(chan NodeResponse[*stream.Message], config.Size()) - } - clientCtx := &CallContext[Req, *emptypb.Empty]{ - Context: ctx, - config: config, - request: req, - method: method, - msgID: config.nextMsgID(), - oneway: true, - replyChan: replyChan, + callCtx := &CallContext[Req, *emptypb.Empty]{ + Context: ctx, + config: config, + request: req, + method: method, + msgID: config.nextMsgID(), + oneway: true, } - clientCtx.responseSeq = clientCtx.defaultResponseSeq() - clientCtx.applyInterceptors(interceptors) - return clientCtx + callCtx.responseSeq = callCtx.defaultResponseSeq() + return callCtx } // ------------------------------------------------------------------------- @@ -192,18 +188,6 @@ func (c *CallContext[Req, Resp]) enqueue(n *Node, msg *stream.Message) { }) } -// applyInterceptors chains the given interceptors, wrapping the response sequence. -// Each interceptor receives the current response sequence and returns a new one. -// Interceptors are applied in order, with each wrapping the previous result. -func (c *CallContext[Req, Resp]) applyInterceptors(interceptors []any) { - responseSeq := c.responseSeq - for _, ic := range interceptors { - interceptor := ic.(ClientInterceptor[Req, Resp]) - responseSeq = interceptor(c, responseSeq) - } - c.responseSeq = responseSeq -} - // send dispatches requests to all nodes. It delegates to sendWithPerNodeTransformation // if any per-node request transformations are registered. Otherwise, it uses sendShared // to marshal the request once and send the same message to all nodes. @@ -307,45 +291,3 @@ func (c *CallContext[Req, Resp]) streamingResponseSeq() ResponseSeq[Resp] { // ------------------------------------------------------------------------- // Interceptors (Middleware) // ------------------------------------------------------------------------- - -// MapRequest returns an interceptor that applies per-node request transformations. -// Multiple interceptors can be chained together, with transforms applied in order. -// -// The fn receives the original request and a node, and returns the transformed -// request to send to that node. If the function returns an invalid message or nil, -// an ErrSkipNode error is sent for that node, indicating it was skipped. -func MapRequest[Req, Resp msg](fn func(Req, *Node) Req) ClientInterceptor[Req, Resp] { - return func(ctx *CallContext[Req, Resp], next ResponseSeq[Resp]) ResponseSeq[Resp] { - if fn != nil { - ctx.reqTransforms = append(ctx.reqTransforms, fn) - } - return next - } -} - -// MapResponse returns an interceptor that applies per-node response transformations. -// -// The fn receives the response from a node and the node itself, and returns the -// transformed response. -func MapResponse[Req, Resp msg](fn func(Resp, *Node) Resp) ClientInterceptor[Req, Resp] { - return func(ctx *CallContext[Req, Resp], next ResponseSeq[Resp]) ResponseSeq[Resp] { - if fn == nil { - return next - } - // Wrap the response iterator with the transformation logic. - return func(yield func(NodeResponse[Resp]) bool) { - for resp := range next { - // We only apply the transformation if there is no error. - // Errors are passed through as-is. - if resp.Err == nil { - if node := ctx.Node(resp.NodeID); node != nil { - resp.Value = fn(resp.Value, node) - } - } - if !yield(resp) { - return - } - } - } - } -} diff --git a/call_quorum_test.go b/call_quorum_test.go index 16c8c6f41..852a762a7 100644 --- a/call_quorum_test.go +++ b/call_quorum_test.go @@ -52,7 +52,7 @@ func checkQuorumCall(t *testing.T, gotErr, wantErr error, expectedNodeErrors ... func TestQuorumCall(t *testing.T) { // type alias short hand for the responses type - type respType = *gorums.Responses[*pb.StringValue] + type respType = *gorums.Call[*pb.StringValue, *pb.StringValue] tests := []struct { name string call func(respType) (*pb.StringValue, error) @@ -110,7 +110,7 @@ func TestQuorumCallPartialFailures(t *testing.T) { const numServers = 3 - type respType = *gorums.Responses[*pb.StringValue] + type respType = *gorums.Call[*pb.StringValue, *pb.StringValue] // Helper to create QuorumCall variants quorumcall := func(name string, aggregateFunc func(respType) (*pb.StringValue, error)) callInfo { @@ -123,12 +123,12 @@ func TestQuorumCallPartialFailures(t *testing.T) { } } - // Helper to create Multicast variants - multicast := func(name string, opts ...gorums.CallOption) callInfo { + // Helper to create a Multicast variant; Wait returns the send error. + multicast := func(name string) callInfo { return callInfo{ name: "Multicast/" + name, callFunc: func(ctx *gorums.ConfigContext, req *pb.StringValue) error { - return gorums.Multicast(ctx, req, mock.TestMethod, opts...) + return gorums.Multicast(ctx, req, mock.TestMethod).Send() }, } } @@ -138,16 +138,12 @@ func TestQuorumCallPartialFailures(t *testing.T) { failing int // number of servers to stop (0 to 3) wantErr error }{ - // Multicast: Fails if ANY node fails + // Multicast Wait: Fails if ANY node fails {multicast("Wait"), 0, nil}, {multicast("Wait"), 1, gorums.ErrSendFailure}, {multicast("Wait"), 2, gorums.ErrSendFailure}, {multicast("Wait"), 3, gorums.ErrSendFailure}, - // Multicast with IgnoreErrors: Should not return error even if nodes fail - {multicast("IgnoreErrors", gorums.IgnoreErrors()), 0, nil}, - {multicast("IgnoreErrors", gorums.IgnoreErrors()), 3, nil}, - // QuorumCall Majority (2/3): Tolerates 1 failure {quorumcall("Majority", respType.Majority), 0, nil}, {quorumcall("Majority", respType.Majority), 1, nil}, diff --git a/call_test.go b/call_test.go new file mode 100644 index 000000000..2a18801e2 --- /dev/null +++ b/call_test.go @@ -0,0 +1,347 @@ +package gorums_test + +import ( + "context" + "errors" + "fmt" + "slices" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/relab/gorums" + "github.com/relab/gorums/gorumstest" + "github.com/relab/gorums/internal/testutils/mock" + pb "google.golang.org/protobuf/types/known/wrapperspb" +) + +// TestOnewayNoResourceLeak verifies that a one-way multicast does not register +// a router entry, so no pending calls are left behind. One-way sends are +// confirmed directly on the reply channel and never round-trip through the +// router, which is what keeps the pending set empty. +func TestOnewayNoResourceLeak(t *testing.T) { + servers := gorumstest.LocalServers(t, 3) + for _, srv := range servers { + srv.RegisterHandler(mock.TestMethod, func(_ gorums.ServerContext, _ *gorums.Message) (*gorums.Message, error) { + return nil, nil + }) + } + for _, srv := range servers { + srv.WaitForPeers(t.Context(), func(cfg gorums.Config) bool { + return cfg.Size() == 3 + }) + } + cfg := servers[0].PeerConfig() + ctx := gorumstest.Context(t, 5*time.Second) + for i := range 1000 { + if err := gorums.Multicast(cfg.Context(ctx), pb.String(fmt.Sprintf("mc-%d", i)), mock.TestMethod).Send(); err != nil { + t.Fatalf("Multicast %d: %v", i, err) + } + } + gorumstest.WaitUntil(t, 5*time.Second, func() bool { + for _, node := range cfg.Nodes() { + if node.PendingCount() > 0 { + return false + } + } + return true + }) + + for _, node := range cfg.Nodes() { + if pc := node.PendingCount(); pc > 0 { + t.Errorf("node %d: pending = %d; expected 0", node.ID(), pc) + } + } +} + +// TestOnewayDroppedHandleDoesNotDispatch verifies that a one-way call handle +// dropped without consuming it never dispatches the request. +func TestOnewayDroppedHandleDoesNotDispatch(t *testing.T) { + var received atomic.Int32 + servers := gorumstest.LocalServers(t, 3) + for _, srv := range servers { + srv.RegisterHandler(mock.TestMethod, func(_ gorums.ServerContext, _ *gorums.Message) (*gorums.Message, error) { + received.Add(1) + return nil, nil + }) + } + for _, srv := range servers { + srv.WaitForPeers(t.Context(), func(cfg gorums.Config) bool { + return cfg.Size() == 3 + }) + } + cfg := servers[0].PeerConfig() + ctx := gorumstest.Context(t, 2*time.Second) + // Drop the handle without consuming it: nothing must be sent. + _ = gorums.Multicast(cfg.Context(ctx), pb.String("dropped"), mock.TestMethod) + // Allow time for any erroneous dispatch to reach the servers. + time.Sleep(200 * time.Millisecond) + if got := received.Load(); got != 0 { + t.Errorf("dropped one-way handle dispatched to %d nodes; want 0", got) + } +} + +// TestOnewayCallDoubleDispatchPanics verifies that consuming the same handle a +// second time panics, in every combination of the two terminals, rather than +// silently re-sending the request or blocking on confirmations the first +// dispatch already drained. +func TestOnewayCallDoubleDispatchPanics(t *testing.T) { + servers := gorumstest.LocalServers(t, 3) + for _, srv := range servers { + srv.RegisterHandler(mock.TestMethod, func(_ gorums.ServerContext, _ *gorums.Message) (*gorums.Message, error) { + return nil, nil + }) + } + for _, srv := range servers { + srv.WaitForPeers(t.Context(), func(cfg gorums.Config) bool { + return cfg.Size() == 3 + }) + } + cfg := servers[0].PeerConfig() + + tests := []struct { + name string + run func(call *gorums.OnewayCall[*pb.StringValue]) + }{ + {"SendAfterSend", func(c *gorums.OnewayCall[*pb.StringValue]) { + _ = c.Send() + _ = c.Send() + }}, + {"AsyncAfterSend", func(c *gorums.OnewayCall[*pb.StringValue]) { + _ = c.Send() + c.Async() + }}, + {"SendAfterAsync", func(c *gorums.OnewayCall[*pb.StringValue]) { + _ = c.Async().Wait() + _ = c.Send() + }}, + {"AsyncAfterAsync", func(c *gorums.OnewayCall[*pb.StringValue]) { + _ = c.Async().Wait() + c.Async() + }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx := gorumstest.Context(t, 2*time.Second) + call := gorums.Multicast(cfg.Context(ctx), pb.String("x"), mock.TestMethod) + defer func() { + if recover() == nil { + t.Errorf("%s did not panic", tt.name) + } + }() + tt.run(call) + }) + } +} + +// TestOnewayCallAsync verifies the two properties that make Async worth having +// over Send: it returns before the sends complete, so a single goroutine can +// keep several calls in flight, and the deferred Wait still reports every send +// error that Send would have reported. +func TestOnewayCallAsync(t *testing.T) { + const calls = 20 + var received atomic.Int32 + servers := gorumstest.LocalServers(t, 3) + for _, srv := range servers { + srv.RegisterHandler(mock.TestMethod, func(_ gorums.ServerContext, _ *gorums.Message) (*gorums.Message, error) { + received.Add(1) + return nil, nil + }) + } + for _, srv := range servers { + srv.WaitForPeers(t.Context(), func(cfg gorums.Config) bool { + return cfg.Size() == 3 + }) + } + cfg := servers[0].PeerConfig() + ctx := gorumstest.Context(t, 5*time.Second) + + // Dispatch every call before collecting any of them: with Send this loop + // could not proceed until each multicast had reached all three nodes. + handles := make([]*gorums.OnewayAsync, calls) + for i := range handles { + handles[i] = gorums.Multicast(cfg.Context(ctx), pb.String(fmt.Sprintf("async-%d", i)), mock.TestMethod).Async() + } + for i, h := range handles { + if err := h.Wait(); err != nil { + t.Errorf("handle %d: Wait = %v, want nil", i, err) + } + // Wait is idempotent: a second call returns the same result. + if err := h.Wait(); err != nil { + t.Errorf("handle %d: second Wait = %v, want nil", i, err) + } + } + gorumstest.WaitUntil(t, 5*time.Second, func() bool { + return received.Load() == calls*int32(cfg.Size()) + }) + if got, want := received.Load(), int32(calls*cfg.Size()); got != want { + t.Errorf("servers received %d messages; want %d", got, want) + } +} + +// TestOnewayCallAsyncReportsSendError verifies that a send failure reaches the +// caller through the deferred Wait, so Async does not reintroduce the silent +// drop that a fire-and-forget terminal had. +func TestOnewayCallAsyncReportsSendError(t *testing.T) { + const numServers = 3 + var stopNodes func(...int) + config := gorumstest.Config(t, numServers, gorumstest.DefaultServer, gorumstest.WithStopFunc(t, &stopNodes)) + ctx := config.Context(t.Context()) + + // Warm up so the streams are established before they are torn down. + if err := gorums.Multicast(ctx, pb.String("warmup"), mock.TestMethod).Send(); err != nil { + t.Fatalf("warmup: %v", err) + } + stopNodes(slices.Collect(gorumstest.Range(numServers))...) + + // Retry until the torn-down streams are observed, as the quorum-call + // failure tests do; the send fails once the stream is gone, not the + // instant the server stops. + var err error + for range 5 { + if err = gorums.Multicast(ctx, pb.String("x"), mock.TestMethod).Async().Wait(); err != nil { + break + } + time.Sleep(10 * time.Millisecond) + } + if err == nil { + t.Fatal("Wait = nil, want a send failure after the servers stopped") + } + if !errors.Is(err, gorums.ErrSendFailure) { + t.Errorf("Wait = %v, want %v", err, gorums.ErrSendFailure) + } +} + +// TestCallInterceptAfterDispatchPanics verifies that calling Intercept after a +// terminal method has started dispatch panics, since interceptors can no longer +// influence the in-flight call. +func TestCallInterceptAfterDispatchPanics(t *testing.T) { + config := gorumstest.Config(t, 3, gorumstest.EchoServerFn) + ctx := gorumstest.Context(t, 5*time.Second) + call := gorums.QuorumCall[*pb.StringValue, *pb.StringValue](config.Context(ctx), pb.String("x"), mock.TestMethod) + if _, err := call.Majority(); err != nil { + t.Fatalf("Majority: %v", err) + } + defer func() { + if recover() == nil { + t.Error("Intercept after dispatch did not panic") + } + }() + call.Intercept(gorums.MapResponse[*pb.StringValue](func(r *pb.StringValue, _ *gorums.Node) *pb.StringValue { return r })) +} + +// TestCallInterceptAfterResultsPanics verifies that calling Intercept after +// Results panics, even though Results itself does not dispatch: without this, +// an interceptor registered on the handle after Results had already been +// called would silently fail to apply to the iterator the caller is holding. +func TestCallInterceptAfterResultsPanics(t *testing.T) { + config := gorumstest.Config(t, 3, gorumstest.EchoServerFn) + ctx := gorumstest.Context(t, 5*time.Second) + call := gorums.QuorumCall[*pb.StringValue, *pb.StringValue](config.Context(ctx), pb.String("x"), mock.TestMethod) + _ = call.Results() + defer func() { + if recover() == nil { + t.Error("Intercept after Results did not panic") + } + }() + call.Intercept(gorums.MapResponse[*pb.StringValue](func(r *pb.StringValue, _ *gorums.Node) *pb.StringValue { return r })) +} + +// TestCallInterceptNilIgnored verifies that nil interceptors are ignored rather +// than causing a panic or affecting the result. +func TestCallInterceptNilIgnored(t *testing.T) { + config := gorumstest.Config(t, 3, gorumstest.EchoServerFn) + ctx := gorumstest.Context(t, 5*time.Second) + resp, err := gorums.QuorumCall[*pb.StringValue, *pb.StringValue](config.Context(ctx), pb.String("test"), mock.TestMethod). + Intercept(nil). + Majority() + if err != nil { + t.Fatalf("Majority: %v", err) + } + if resp.GetValue() != "echo: test" { + t.Errorf("got %q, want %q", resp.GetValue(), "echo: test") + } +} + +func TestRemoteCallSuccess(t *testing.T) { + node := gorumstest.Node(t, gorumstest.DefaultServer) + + ctx := gorumstest.Context(t, 5*time.Second) + nodeCtx := node.Context(ctx) + response, err := gorums.RemoteCall[*pb.StringValue, *pb.StringValue](nodeCtx, pb.String(""), mock.TestMethod) + if err != nil { + t.Fatalf("Unexpected error, got: %v, want: %v", err, nil) + } + if response == nil { + t.Fatalf("Unexpected response, got: %v, want: non-nil", nil) + } +} + +func TestRemoteCallDownedNode(t *testing.T) { + node := gorumstest.Node(t, gorumstest.DefaultServer, gorumstest.WithPreConnect(t, func(stopServers func()) { + stopServers() + time.Sleep(300 * time.Millisecond) // wait for servers to fully stop + })) + + ctx := gorumstest.Context(t, 5*time.Second) + nodeCtx := node.Context(ctx) + response, err := gorums.RemoteCall[*pb.StringValue, *pb.StringValue](nodeCtx, pb.String(""), mock.TestMethod) + if err == nil { + t.Fatalf("Expected error, got: %v, want: %v", err, fmt.Errorf("rpc error: code = Unavailable desc = stream is down")) + } + if response != nil { + t.Fatalf("Unexpected response, got: %v, want: %v", response, nil) + } +} + +func TestRemoteCallTimedOut(t *testing.T) { + node := gorumstest.Node(t, gorumstest.DefaultServer) + + ctx, cancel := context.WithTimeout(t.Context(), 0*time.Second) + time.Sleep(50 * time.Millisecond) + defer cancel() + nodeCtx := node.Context(ctx) + response, err := gorums.RemoteCall[*pb.StringValue, *pb.StringValue](nodeCtx, pb.String(""), mock.TestMethod) + if err == nil { + t.Fatalf("Expected error, got: %v, want: %v", err, fmt.Errorf("context deadline exceeded")) + } + if response != nil { + t.Fatalf("Unexpected response, got: %v, want: %v", response, nil) + } +} + +func TestRemoteCallTypeMismatch(t *testing.T) { + node := gorumstest.Node(t, gorumstest.DefaultServer) + + ctx := gorumstest.Context(t, 5*time.Second) + nodeCtx := node.Context(ctx) + response, err := gorums.RemoteCall[*pb.StringValue, *pb.Int32Value](nodeCtx, pb.String(""), mock.TestMethod) + if err != gorums.ErrTypeMismatch { + t.Fatalf("Expected error, got: %v, want: %v", err, gorums.ErrTypeMismatch) + } + if response != nil { + t.Fatalf("Unexpected response, got: %v, want: %v", response, nil) + } +} + +func TestRemoteCallConcurrentAccess(t *testing.T) { + node := gorumstest.Node(t, gorumstest.DefaultServer) + + concurrency := 10 + errCh := make(chan error, concurrency) + var wg sync.WaitGroup + for range concurrency { + wg.Go(func() { + _, err := gorums.RemoteCall[*pb.StringValue, *pb.StringValue](node.Context(t.Context()), pb.String(""), mock.TestMethod) + if err != nil { + errCh <- err + } + }) + } + wg.Wait() + close(errCh) + for err := range errCh { + t.Error(err) + } +} diff --git a/callopts.go b/callopts.go deleted file mode 100644 index 8488e9ad3..000000000 --- a/callopts.go +++ /dev/null @@ -1,50 +0,0 @@ -package gorums - -import ( - "google.golang.org/protobuf/proto" -) - -type callOptions struct { - ignoreErrors bool - interceptors []any // Type-erased interceptors, restored by QuorumCall -} - -// CallOption is a function that sets a value in the given callOptions struct -type CallOption func(*callOptions) - -func getCallOptions(opts ...CallOption) callOptions { - o := callOptions{ - ignoreErrors: false, // default: return error and wait for send completion - } - for _, opt := range opts { - opt(&o) - } - return o -} - -// IgnoreErrors ignores send errors from Unicast or Multicast methods and -// returns immediately instead of blocking until the message has been sent. -// By default, Unicast and Multicast methods return an error if the message -// could not be sent or the context was canceled. -func IgnoreErrors() CallOption { - return func(o *callOptions) { - o.ignoreErrors = true - } -} - -// Interceptors returns a CallOption that adds quorum call interceptors. -// Interceptors are executed in the order provided, modifying the Responses -// object before the user calls a terminal method. -// -// Example: -// -// resp, err := ReadQC(ctx, req, -// gorums.Interceptors(loggingInterceptor, filterInterceptor), -// ).Majority() -func Interceptors[Req, Resp proto.Message](interceptors ...ClientInterceptor[Req, Resp]) CallOption { - return func(o *callOptions) { - for _, interceptor := range interceptors { - o.interceptors = append(o.interceptors, interceptor) - } - } -} diff --git a/callopts_test.go b/callopts_test.go deleted file mode 100644 index e7aa5e625..000000000 --- a/callopts_test.go +++ /dev/null @@ -1,135 +0,0 @@ -package gorums - -import ( - "context" - "fmt" - "testing" - "time" - - "github.com/relab/gorums/internal/testutils/mock" - "go.uber.org/goleak" - "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" - pb "google.golang.org/protobuf/types/known/wrapperspb" -) - -// testLocalServers returns n started Gorums servers forming a symmetric peer -// group on random localhost ports. It is the in-package counterpart of -// gorumstest.LocalServers, which this file cannot use: gorumstest imports -// gorums, so importing it from package gorums's own tests would create an -// import cycle. -func testLocalServers(t testing.TB, n int) []*Server { - t.Helper() - if _, ok := t.(*testing.B); !ok { - t.Cleanup(func() { goleak.VerifyNone(t) }) - } - srvs, stop, err := NewLocalServers(n, WithLocalDialOptions( - WithGRPCDialOptions(grpc.WithTransportCredentials(insecure.NewCredentials())), - )) - if err != nil { - t.Fatal(err) - } - t.Cleanup(stop) - for _, srv := range srvs { - go srv.ListenAndServe() - } - return srvs -} - -// testWaitUntil polls predicate until it returns true or timeout elapses. -// It is the in-package counterpart of gorumstest.WaitUntil. -func testWaitUntil(t testing.TB, timeout time.Duration, predicate func() bool) bool { - t.Helper() - if predicate() { - return true - } - ctx, cancel := context.WithTimeout(t.Context(), timeout) - defer cancel() - ticker := time.NewTicker(10 * time.Millisecond) - defer ticker.Stop() - for { - select { - case <-ctx.Done(): - return predicate() - case <-ticker.C: - if predicate() { - return true - } - } - } -} - -func TestCallOptionsIgnoreErrors(t *testing.T) { - tests := []struct { - name string - callOpts callOptions - wantIgnoreErrors bool - }{ - {name: "Default", callOpts: getCallOptions(), wantIgnoreErrors: false}, - {name: "IgnoreErrors", callOpts: getCallOptions(IgnoreErrors()), wantIgnoreErrors: true}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - if got := tt.callOpts.ignoreErrors; got != tt.wantIgnoreErrors { - t.Errorf("ignoreErrors = %v, want %v", got, tt.wantIgnoreErrors) - } - }) - } -} - -func TestCallOptionsIgnoreErrorsResourceLeak(t *testing.T) { - // Previously leaked because fire-and-forget multicast still registered in router. - // Now fixed: no replyChan → no ResponseChan → no Register. - servers := testLocalServers(t, 3) - for _, srv := range servers { - srv.RegisterHandler(mock.TestMethod, func(_ ServerContext, _ *Message) (*Message, error) { - return nil, nil - }) - } - for _, srv := range servers { - srv.WaitForPeers(t.Context(), func(cfg Config) bool { - return cfg.Size() == 3 - }) - } - cfg := servers[0].PeerConfig() - ctx := testTimeoutContext(t, 5*time.Second) - for i := range 1000 { - Multicast(cfg.Context(ctx), pb.String(fmt.Sprintf("mc-%d", i)), mock.TestMethod, IgnoreErrors()) - } - testWaitUntil(t, 5*time.Second, func() bool { - for _, node := range cfg.Nodes() { - if node.PendingCount() > 0 { - return false - } - } - return true - }) - - for _, node := range cfg.Nodes() { - if pc := node.PendingCount(); pc > 0 { - t.Errorf("node %d: pending = %d; expected 0", node.ID(), pc) - } - } -} - -func BenchmarkGetCallOptions(b *testing.B) { - interceptor := func(_ *CallContext[msg, msg], next ResponseSeq[msg]) ResponseSeq[msg] { return next } - tests := []struct { - numOpts int - }{ - {0}, {1}, {2}, {3}, {4}, {5}, - } - - for _, tc := range tests { - opts := make([]CallOption, tc.numOpts) - for i := range tc.numOpts { - opts[i] = Interceptors(interceptor) - } - b.Run(fmt.Sprintf("options=%d", tc.numOpts), func(b *testing.B) { - b.ReportAllocs() - for b.Loop() { - _ = getCallOptions(opts...) - } - }) - } -} diff --git a/cmd/protoc-gen-gorums/gengorums/template_multicast.go b/cmd/protoc-gen-gorums/gengorums/template_multicast.go index 055550c37..1dc6056c6 100644 --- a/cmd/protoc-gen-gorums/gengorums/template_multicast.go +++ b/cmd/protoc-gen-gorums/gengorums/template_multicast.go @@ -4,7 +4,7 @@ var mcVar = ` {{$genFile := .GenFile}} {{$configContext := "ConfigContext"}} {{$multicast := use "gorums.Multicast" .GenFile}} -{{$callOpt := use "gorums.CallOption" .GenFile}} +{{$onewayCall := use "gorums.OnewayCall" .GenFile}} ` var multicastComment = ` @@ -13,16 +13,22 @@ var multicastComment = ` {{$comments -}} {{else}} // {{$method}} is a multicast call invoked on all nodes in the configuration in ctx. -// Use gorums.MapRequest to send different messages to each node. No replies are collected. +// It returns a one-way call handle; call Send to block until every send +// completes and observe any send errors, or Async to dispatch without waiting. +// Use gorums.MapRequest to send different messages to each node. {{end -}} +// +// Example: +// err := {{$method}}(ctx, in).Send() +// h := {{$method}}(ctx, in).Async(); err := h.Wait() ` var multicastSignature = `func {{$method}}(` + - `ctx *{{$configContext}}, in *{{$in}}, ` + - `opts ...{{$callOpt}}) error { + `ctx *{{$configContext}}, in *{{$in}})` + + ` *{{$onewayCall}}[*{{$in}}] { ` -var multicastBody = ` return {{$multicast}}(ctx, in, "{{$fullName}}", opts...) +var multicastBody = ` return {{$multicast}}(ctx, in, "{{$fullName}}") } ` diff --git a/cmd/protoc-gen-gorums/gengorums/template_quorumcall.go b/cmd/protoc-gen-gorums/gengorums/template_quorumcall.go index 6c7678075..353e3a856 100644 --- a/cmd/protoc-gen-gorums/gengorums/template_quorumcall.go +++ b/cmd/protoc-gen-gorums/gengorums/template_quorumcall.go @@ -28,19 +28,16 @@ var quorumCallVariables = ` {{$configContext := "ConfigContext"}} {{$quorumCall := use "gorums.QuorumCall" .GenFile}} {{$quorumCallStream := use "gorums.QuorumCallStream" .GenFile}} -{{$responses := use "gorums.Responses" .GenFile}} -{{$callOption := use "gorums.CallOption" .GenFile}} +{{$call := use "gorums.Call" .GenFile}} ` var quorumCallSignature = `func {{$method}}(` + - `ctx *{{$configContext}}, in *{{$in}}, ` + - `opts ...{{$callOption}})` + - ` *{{$responses}}[*{{$out}}] { + `ctx *{{$configContext}}, in *{{$in}})` + + ` *{{$call}}[*{{$in}}, *{{$out}}] { ` var quorumCallBody = ` return {{$quorumCall}}[*{{$in}}, *{{$out}}]( ctx, in, "{{$fullName}}", - opts..., ) } ` @@ -69,7 +66,6 @@ var quorumCallStreamComment = ` var quorumCallStreamBody = ` return {{$quorumCallStream}}[*{{$in}}, *{{$out}}]( ctx, in, "{{$fullName}}", - opts..., ) } ` diff --git a/cmd/protoc-gen-gorums/gengorums/template_unicast.go b/cmd/protoc-gen-gorums/gengorums/template_unicast.go index 9eb0442a1..af6630327 100644 --- a/cmd/protoc-gen-gorums/gengorums/template_unicast.go +++ b/cmd/protoc-gen-gorums/gengorums/template_unicast.go @@ -4,7 +4,7 @@ var unicastVar = ` {{$genFile := .GenFile}} {{$nodeContext := "NodeContext"}} {{$unicast := use "gorums.Unicast" .GenFile}} -{{$callOpt := use "gorums.CallOption" .GenFile}} +{{$onewayCall := use "gorums.OnewayCall" .GenFile}} ` var unicastComment = ` @@ -12,16 +12,23 @@ var unicastComment = ` {{if ne $comments ""}} {{$comments -}} {{else}} -// {{$method}} is a unicast call invoked on the node in ctx. -// No reply is returned to the client. +// {{$method}} is a unicast call invoked on the node in ctx; no reply is +// returned to the client. It returns a one-way call handle; call Send to block +// until the send completes and observe any send error, or Async to dispatch +// without waiting. {{end -}} +// +// Example: +// err := {{$method}}(ctx, in).Send() +// h := {{$method}}(ctx, in).Async(); err := h.Wait() ` var unicastSignature = `func {{$method}}(` + - `ctx *{{$nodeContext}}, in *{{$in}}, opts ...{{$callOpt}}) error { + `ctx *{{$nodeContext}}, in *{{$in}})` + + ` *{{$onewayCall}}[*{{$in}}] { ` -var unicastBody = ` return {{$unicast}}(ctx, in, "{{$fullName}}", opts...) +var unicastBody = ` return {{$unicast}}(ctx, in, "{{$fullName}}") } ` diff --git a/correctable.go b/correctable.go index 58e5da877..25bc84d47 100644 --- a/correctable.go +++ b/correctable.go @@ -53,8 +53,10 @@ func (c *Correctable[Resp]) Watch(level int) <-chan struct{} { return ch } -// Correctable returns a Correctable that provides progressive updates -// as responses arrive. The level increases with each successful response. +// Correctable returns a call result that updates as responses arrive. +// Its level increases with each successful response and completes at threshold. +// A response skipped by a request transform ([ErrSkipNode]) counts toward +// neither the level nor the final node-error count. // Use this for correctable quorum patterns where you want to observe // intermediate states. // @@ -65,6 +67,10 @@ func (c *Correctable[Resp]) Watch(level int) <-chan struct{} { // <-corr.Watch(2) // resp, level, err := corr.Get() func (r *Responses[Resp]) Correctable(threshold int) *Correctable[Resp] { + // Mark dispatched before spawning the goroutine so a later Intercept panics + // deterministically; the actual send is triggered lazily by ranging r.seq. + r.markDispatched() + corr := &Correctable[Resp]{ level: LevelNotSet, donech: make(chan struct{}, 1), diff --git a/examples/storage/client.go b/examples/storage/client.go index d8307d1b6..85a5cf92b 100644 --- a/examples/storage/client.go +++ b/examples/storage/client.go @@ -26,7 +26,7 @@ func runClient(addresses []string) error { // newestValue processes responses from a ReadQC call and returns the reply // with the most recent timestamp. -func newestValue(responses *gorums.Responses[*proto.ReadResponse]) (*proto.ReadResponse, error) { +func newestValue(responses *gorums.Call[*proto.ReadRequest, *proto.ReadResponse]) (*proto.ReadResponse, error) { var newest *proto.ReadResponse for resp := range responses.Results() { if resp.Err != nil { @@ -44,7 +44,7 @@ func newestValue(responses *gorums.Responses[*proto.ReadResponse]) (*proto.ReadR // numUpdated processes responses from a WriteQC call and returns true if // a majority of nodes updated their value. -func numUpdated(responses *gorums.Responses[*proto.WriteResponse]) (*proto.WriteResponse, error) { +func numUpdated(responses *gorums.Call[*proto.WriteRequest, *proto.WriteResponse]) (*proto.WriteResponse, error) { var count int size := responses.Size() for resp := range responses.Results() { diff --git a/examples/storage/repl.go b/examples/storage/repl.go index 39beac707..0d149913f 100644 --- a/examples/storage/repl.go +++ b/examples/storage/repl.go @@ -197,7 +197,7 @@ func (r repl) unicast(args []string) { nodeCtx := node.Context(ctx) err = pb.WriteUnicast(nodeCtx, pb.WriteRequest_builder{ Key: args[1], Value: args[2], Time: timestamppb.Now(), - }.Build()) + }.Build()).Send() cancel() if err != nil { fmt.Printf("Write unicast failed to send: %v\n", err) @@ -217,7 +217,7 @@ func (r repl) multicast(args []string) { cfgCtx := r.cfg.Context(ctx) err := pb.WriteMulticast(cfgCtx, pb.WriteRequest_builder{ Key: args[0], Value: args[1], Time: timestamppb.Now(), - }.Build()) + }.Build()).Send() cancel() if err != nil { fmt.Printf("Write multicast failed to send: %v\n", err) diff --git a/examples/storage/server.go b/examples/storage/server.go index acac58bb6..802fad08e 100644 --- a/examples/storage/server.go +++ b/examples/storage/server.go @@ -234,7 +234,7 @@ func (s *storageServer) WriteNestedMulticast(ctx gorums.ServerContext, req *pb.W } // Release before nested outbound calls to avoid blocking inbound recv processing. ctx.Release() - if err := pb.WriteMulticast(cfg.Context(ctx), req); err != nil { + if err := pb.WriteMulticast(cfg.Context(ctx), req).Send(); err != nil { return nil, fmt.Errorf("write_nested_multicast: %w", err) } return pb.WriteResponse_builder{New: true}.Build(), nil diff --git a/inbound_manager_test.go b/inbound_manager_test.go index 1c0bb039f..95509c1d7 100644 --- a/inbound_manager_test.go +++ b/inbound_manager_test.go @@ -881,14 +881,21 @@ func TestClientConfigMixedMode(t *testing.T) { } } -// TestClientConfigServerCallsClient verifies that a server dispatches a reverse-direction +// TestConnectedClientsServerCallsClient verifies that a server dispatches a reverse-direction // multicast to a connected client via [ServerContext.ConnectedClients]. -func TestClientConfigServerCallsClient(t *testing.T) { +func TestConnectedClientsServerCallsClient(t *testing.T) { // Register the server handler before starting so it is present before clients arrive. srv := NewServer() srv.RegisterHandler(mock.TestMethod, func(ctx ServerContext, _ *Message) (*Message, error) { - if clients := ctx.ConnectedClients(); len(clients) > 0 { - _ = Multicast(clients.Context(ctx), pb.String("ping"), mock.Stream) + if cfg := ctx.ConnectedClients(); len(cfg) > 0 { + // Release before the back-channel send: Send blocks until every + // client's send completes, and holding the dispatch lock across + // that wait would stop this connection from reading further + // inbound frames. + ctx.Release() + if err := Multicast(cfg.Context(ctx), pb.String("ping"), mock.Stream).Send(); err != nil { + t.Errorf("back-channel Multicast: %v", err) + } } return nil, nil // one-way }) @@ -897,7 +904,7 @@ func TestClientConfigServerCallsClient(t *testing.T) { var wg sync.WaitGroup wg.Add(1) - // Client: a Server whose reverse-direction mock.Stream handler is wired in via WithServer. + // Client: a Server whose reverse-direction mock.Stream handler is wired in via WithBackChannel. clientSrv := NewServer() clientSrv.RegisterHandler(mock.Stream, func(_ ServerContext, _ *Message) (*Message, error) { wg.Done() @@ -914,7 +921,7 @@ func TestClientConfigServerCallsClient(t *testing.T) { // Trigger: client multicasts TestMethod to the server; server fans it back via ClientConfig. ctx := testTimeoutContext(t, 2*time.Second) - if err := Multicast(clientConfig.Context(ctx), pb.String("trigger"), mock.TestMethod); err != nil { + if err := Multicast(clientConfig.Context(ctx), pb.String("trigger"), mock.TestMethod).Send(); err != nil { t.Fatalf("Multicast error: %v", err) } diff --git a/internal/tests/oneway/oneway_test.go b/internal/tests/oneway/oneway_test.go index 529593f4d..ab05a62fe 100644 --- a/internal/tests/oneway/oneway_test.go +++ b/internal/tests/oneway/oneway_test.go @@ -4,8 +4,8 @@ import ( context "context" "fmt" "slices" - "sync" "testing" + "time" "github.com/relab/gorums" "github.com/relab/gorums/gorumstest" @@ -15,10 +15,17 @@ import ( const numCalls = 50 +// recvTimeout bounds how long a subtest waits for the messages it sent. A send +// that succeeded still leaves the server free to drop or delay the message, so +// an unbounded wait would hang the package until the test binary's timeout +// instead of reporting the shortfall. +const recvTimeout = 10 * time.Second + type onewaySrv struct { benchmark bool - wg sync.WaitGroup - received chan *oneway.Request + // received buffers the messages of one subtest, with headroom so that a + // straggler arriving after a failed subtest cannot block a handler. + received chan *oneway.Request } func (s *onewaySrv) Unicast(_ gorums.ServerContext, r *oneway.Request) { @@ -26,7 +33,6 @@ func (s *onewaySrv) Unicast(_ gorums.ServerContext, r *oneway.Request) { return } s.received <- r - s.wg.Done() } func (s *onewaySrv) Multicast(_ gorums.ServerContext, r *oneway.Request) { @@ -34,88 +40,127 @@ func (s *onewaySrv) Multicast(_ gorums.ServerContext, r *oneway.Request) { return } s.received <- r - s.wg.Done() +} + +// cluster is a set of servers and the configuration addressing them, shared by +// every subtest of a table that needs that many servers. +type cluster struct { + cfg oneway.Config + srvs []*onewaySrv +} + +// reset discards messages left over from an earlier subtest so the next one +// starts from a known state. A subtest that received everything it sent leaves +// nothing behind. +func (c *cluster) reset() { + for _, srv := range c.srvs { + for range len(srv.received) { + <-srv.received + } + } +} + +// received returns the messages that the given server received, sorted by +// their Num field. Sorting avoids flakiness from multicast reordering. If +// fewer than want messages arrive within [recvTimeout] it reports the +// shortfall and returns nil, since every later message then compares against +// the wrong expected value; a dropped one-way message surfaces this way. +func (c *cluster) received(t *testing.T, i, want int) []uint64 { + t.Helper() + got := gorumstest.Collect(t, recvTimeout, want, c.srvs[i].received) + if len(got) != want { + t.Errorf("server %d received %d messages, expected %d", i, len(got), want) + return nil + } + nums := make([]uint64, len(got)) + for j, r := range got { + nums[j] = r.GetNum() + } + slices.Sort(nums) + return nums +} + +// clusters returns a lookup that lazily creates one shared cluster per +// configuration size and resets it before each use. Sharing clusters across +// the subtests of a table keeps the number of connections proportional to the +// distinct sizes rather than to the number of subtests: with real TCP +// listeners, every subtest would otherwise leave one socket per node in +// TIME_WAIT, and a high -count run can exhaust the ephemeral port range. +// +// The lookup must be called from the goroutine running t, not from a subtest, +// since it registers servers and cleanup on t. Only the first cluster +// registers a goroutine leak check: cleanup functions run in reverse +// registration order, so that check runs after every cluster has been torn +// down, whereas a check registered by a later cluster would run while earlier +// clusters are still serving. +func clusters(t *testing.T) func(cfgSize int) *cluster { + cache := make(map[int]*cluster) + return func(cfgSize int) *cluster { + t.Helper() + if c, ok := cache[cfgSize]; ok { + c.reset() + return c + } + var opts []gorumstest.Option + if len(cache) > 0 { + opts = append(opts, gorumstest.SkipGoleak()) + } + cfg, srvs := setupWithNodeMap(t, cfgSize, opts...) + c := &cluster{cfg: cfg, srvs: srvs} + cache[cfgSize] = c + return c + } } // setupWithNodeMap sets up servers and configuration with sequential node IDs // (1, 2, 3, ...) matching the server array indices. This is needed for tests like // TestMulticastPerNode that verify per-node message transformations based on node ID. -func setupWithNodeMap(t testing.TB, cfgSize int) (cfg oneway.Config, srvs []*onewaySrv) { +func setupWithNodeMap(t testing.TB, cfgSize int, opts ...gorumstest.Option) (cfg oneway.Config, srvs []*onewaySrv) { t.Helper() srvs = make([]*onewaySrv, cfgSize) for i := range cfgSize { - srvs[i] = &onewaySrv{received: make(chan *oneway.Request, numCalls)} + srvs[i] = &onewaySrv{received: make(chan *oneway.Request, 2*numCalls)} } cfg = gorumstest.Config(t, cfgSize, func(i int) gorums.ServerIface { srv := gorums.NewServer() oneway.RegisterOnewayTestServer(srv, srvs[i]) return srv - }) + }, opts...) return cfg, srvs } func TestOnewayCalls(t *testing.T) { tests := []struct { - name string - calls int - servers int - sendWait bool + name string + calls int + servers int + unicast bool }{ - {name: "UnicastSendWaiting____", calls: numCalls, servers: 1, sendWait: true}, - {name: "UnicastNoSendWaiting__", calls: numCalls, servers: 1, sendWait: false}, - {name: "MulticastSendWaiting__", calls: numCalls, servers: 1, sendWait: true}, - {name: "MulticastNoSendWaiting", calls: numCalls, servers: 1, sendWait: false}, - {name: "MulticastSendWaiting__", calls: numCalls, servers: 3, sendWait: true}, - {name: "MulticastNoSendWaiting", calls: numCalls, servers: 3, sendWait: false}, - {name: "MulticastSendWaiting__", calls: numCalls, servers: 9, sendWait: true}, - {name: "MulticastNoSendWaiting", calls: numCalls, servers: 9, sendWait: false}, + {name: "Unicast__", calls: numCalls, servers: 1, unicast: true}, + {name: "Multicast", calls: numCalls, servers: 1}, + {name: "Multicast", calls: numCalls, servers: 3}, + {name: "Multicast", calls: numCalls, servers: 9}, } + newCluster := clusters(t) for _, test := range tests { + c := newCluster(test.servers) t.Run(fmt.Sprintf("%s/Servers=%d", test.name, test.servers), func(t *testing.T) { - config, srvs := setupWithNodeMap(t, test.servers) - for i := range srvs { - srvs[i].wg.Add(test.calls) - } - for c := 1; c <= test.calls; c++ { - in := oneway.Request_builder{Num: uint64(c)}.Build() - if config.Size() == 1 { - node := config[0] - nodeCtx := node.Context(context.Background()) - if test.sendWait { - if err := oneway.Unicast(nodeCtx, in); err != nil { - t.Error(err) - } - } else { - if err := oneway.Unicast(nodeCtx, in, gorums.IgnoreErrors()); err != nil { - t.Error(err) - } - } + for i := 1; i <= test.calls; i++ { + in := oneway.Request_builder{Num: uint64(i)}.Build() + var err error + if test.unicast { + err = oneway.Unicast(c.cfg[0].Context(context.Background()), in).Send() } else { - cfgCtx := config.Context(context.Background()) - if test.sendWait { - if err := oneway.Multicast(cfgCtx, in); err != nil { - t.Error(err) - } - } else { - if err := oneway.Multicast(cfgCtx, in, gorums.IgnoreErrors()); err != nil { - t.Error(err) - } - } + err = oneway.Multicast(c.cfg.Context(context.Background()), in).Send() + } + if err != nil { + t.Error(err) } } // Check that each server received expected oneway messages - for i := range srvs { - srvs[i].wg.Wait() - close(srvs[i].received) - received := make([]uint64, 0, test.calls) - for r := range srvs[i].received { - received = append(received, r.GetNum()) - } - // Sort received messages to avoid test flakiness - // due to message reordering in multicast tests - slices.Sort(received) - for j, got := range received { + for i := range c.srvs { + for j, got := range c.received(t, i, test.calls) { want := uint64(j + 1) if want != got { t.Errorf("%s: received[%d] = %d, expected %d", test.name, j, got, want) @@ -149,74 +194,40 @@ func TestMulticastPerNode(t *testing.T) { name string calls int servers int - sendWait bool ignoreNodes []uint32 }{ - {name: "MulticastPerNodeNoSendWaiting", calls: numCalls, servers: 1, sendWait: false}, - {name: "MulticastPerNodeNoSendWaiting", calls: numCalls, servers: 3, sendWait: false}, - {name: "MulticastPerNodeNoSendWaiting", calls: numCalls, servers: 9, sendWait: false}, - {name: "MulticastPerNodeSendWaiting", calls: numCalls, servers: 1, sendWait: true}, - {name: "MulticastPerNodeSendWaiting", calls: numCalls, servers: 3, sendWait: true}, - {name: "MulticastPerNodeSendWaiting", calls: numCalls, servers: 9, sendWait: true}, - {name: "MulticastPerNodeNoSendWaitingIgnoreNodes", calls: numCalls, servers: 3, sendWait: false, ignoreNodes: []uint32{0}}, - {name: "MulticastPerNodeNoSendWaitingIgnoreNodes", calls: numCalls, servers: 3, sendWait: false, ignoreNodes: []uint32{1}}, - {name: "MulticastPerNodeNoSendWaitingIgnoreNodes", calls: numCalls, servers: 3, sendWait: false, ignoreNodes: []uint32{0, 1}}, - {name: "MulticastPerNodeNoSendWaitingIgnoreNodes", calls: numCalls, servers: 3, sendWait: false, ignoreNodes: []uint32{0, 1, 2}}, - {name: "MulticastPerNodeSendWaitingIgnoreNodes", calls: numCalls, servers: 3, sendWait: true, ignoreNodes: []uint32{0}}, - {name: "MulticastPerNodeSendWaitingIgnoreNodes", calls: numCalls, servers: 3, sendWait: true, ignoreNodes: []uint32{1}}, - {name: "MulticastPerNodeSendWaitingIgnoreNodes", calls: numCalls, servers: 3, sendWait: true, ignoreNodes: []uint32{0, 1}}, - {name: "MulticastPerNodeSendWaitingIgnoreNodes", calls: numCalls, servers: 3, sendWait: true, ignoreNodes: []uint32{0, 1, 2}}, + {name: "MulticastPerNode", calls: numCalls, servers: 1}, + {name: "MulticastPerNode", calls: numCalls, servers: 3}, + {name: "MulticastPerNode", calls: numCalls, servers: 9}, + {name: "MulticastPerNodeIgnoreNodes", calls: numCalls, servers: 3, ignoreNodes: []uint32{0}}, + {name: "MulticastPerNodeIgnoreNodes", calls: numCalls, servers: 3, ignoreNodes: []uint32{1}}, + {name: "MulticastPerNodeIgnoreNodes", calls: numCalls, servers: 3, ignoreNodes: []uint32{0, 1}}, + {name: "MulticastPerNodeIgnoreNodes", calls: numCalls, servers: 3, ignoreNodes: []uint32{0, 1, 2}}, } + newCluster := clusters(t) for _, test := range tests { + c := newCluster(test.servers) t.Run(fmt.Sprintf("%s/Servers=%d/IgnoredNodes=%v", test.name, test.servers, test.ignoreNodes), func(t *testing.T) { - config, srvs := setupWithNodeMap(t, test.servers) - nodeIDs := config.NodeIDs() + nodeIDs := c.cfg.NodeIDs() // create a test-local ignore function to avoid data races between tests ignore := makeIgnoreFunc(test.ignoreNodes) mapFunc := makeMapFunc(ignore) - for i := range srvs { - if ignore(nodeIDs[i]) { - continue // don't check ignored nodes - } - srvs[i].wg.Add(test.calls) - } - - for c := 1; c <= test.calls; c++ { - in := oneway.Request_builder{Num: uint64(c)}.Build() - cfgCtx := config.Context(context.Background()) + for i := 1; i <= test.calls; i++ { + in := oneway.Request_builder{Num: uint64(i)}.Build() + cfgCtx := c.cfg.Context(context.Background()) mapInterceptor := gorums.MapRequest[*oneway.Request, *emptypb.Empty](mapFunc) - if test.sendWait { - if err := oneway.Multicast(cfgCtx, in, - gorums.Interceptors(mapInterceptor), - ); err != nil { - t.Error(err) - } - } else { - if err := oneway.Multicast(cfgCtx, in, - gorums.Interceptors(mapInterceptor), - gorums.IgnoreErrors(), - ); err != nil { - t.Error(err) - } + if err := oneway.Multicast(cfgCtx, in).Intercept(mapInterceptor).Send(); err != nil { + t.Error(err) } } // Check that each server received expected oneway messages - for i := range srvs { + for i := range c.srvs { if ignore(nodeIDs[i]) { continue // don't check ignored nodes } - srvs[i].wg.Wait() - close(srvs[i].received) - received := make([]uint64, 0, test.calls) - for r := range srvs[i].received { - received = append(received, r.GetNum()) - } - // Sort received messages to avoid test flakiness - // due to message reordering in multicast tests - slices.Sort(received) - for j, got := range received { + for j, got := range c.received(t, i, test.calls) { want := add(uint64(j+1), nodeIDs[i]) if want != got { t.Errorf("%s: received[%d] = %d, expected %d, nodeID=%d", test.name, j, got, want, nodeIDs[i]) @@ -234,20 +245,13 @@ func BenchmarkUnicast(b *testing.B) { } node := cfg[0] in := oneway.Request_builder{Num: 0}.Build() - b.Run("UnicastSendWaiting__", func(b *testing.B) { - for c := 1; c <= b.N; c++ { - in.SetNum(uint64(c)) - nodeCtx := node.Context(context.Background()) - oneway.Unicast(nodeCtx, in) + for c := 1; c <= b.N; c++ { + in.SetNum(uint64(c)) + nodeCtx := node.Context(context.Background()) + if err := oneway.Unicast(nodeCtx, in).Send(); err != nil { + b.Fatal(err) } - }) - b.Run("UnicastNoSendWaiting", func(b *testing.B) { - for c := 1; c <= b.N; c++ { - in.SetNum(uint64(c)) - nodeCtx := node.Context(context.Background()) - oneway.Unicast(nodeCtx, in, gorums.IgnoreErrors()) - } - }) + } } func BenchmarkMulticast(b *testing.B) { @@ -256,18 +260,11 @@ func BenchmarkMulticast(b *testing.B) { srv.benchmark = true } in := oneway.Request_builder{Num: 0}.Build() - b.Run("MulticastSendWaiting__", func(b *testing.B) { - for c := 1; c <= b.N; c++ { - in.SetNum(uint64(c)) - cfgCtx := config.Context(context.Background()) - oneway.Multicast(cfgCtx, in) + for c := 1; c <= b.N; c++ { + in.SetNum(uint64(c)) + cfgCtx := config.Context(context.Background()) + if err := oneway.Multicast(cfgCtx, in).Send(); err != nil { + b.Fatal(err) } - }) - b.Run("MulticastNoSendWaiting", func(b *testing.B) { - for c := 1; c <= b.N; c++ { - in.SetNum(uint64(c)) - cfgCtx := config.Context(context.Background()) - oneway.Multicast(cfgCtx, in, gorums.IgnoreErrors()) - } - }) + } } diff --git a/multicast.go b/multicast.go index 9e12ed0af..6aa750005 100644 --- a/multicast.go +++ b/multicast.go @@ -1,48 +1,18 @@ package gorums -import ( - "errors" -) +import "google.golang.org/protobuf/proto" -// Multicast is a one-way call; no replies are returned to the client. +// Multicast is a one-way call to every node in the configuration; no replies +// are returned to the client. It returns an [OnewayCall] handle that dispatches +// the request only when it is consumed: [OnewayCall.Send] blocks until the send +// completes for every node and reports any send failures, while +// [OnewayCall.Async] dispatches without waiting and defers those failures to +// [OnewayAsync.Wait]. // -// By default, this method blocks until messages have been sent to all nodes. -// This ensures that send operations complete before the caller proceeds, which can -// be useful for observing context cancellation or for pacing message sends. -// If the sending fails, the error is returned to the caller. +// Register per-node request transforms with [OnewayCall.Intercept] and +// [MapRequest] before consuming the handle. // -// With the IgnoreErrors call option, the method returns nil immediately after -// enqueueing messages to all nodes (fire-and-forget semantics). -// -// Multicast supports request transformation interceptors via the gorums.Interceptors -// option. Use gorums.MapRequest to transform requests per-node. -// -// This method should be used by generated code only. -func Multicast[Req msg](ctx *ConfigContext, req Req, method string, opts ...CallOption) error { - callOpts := getCallOptions(opts...) - waitForSend := !callOpts.ignoreErrors - - clientCtx := newMulticastCallContext(ctx, req, method, waitForSend, callOpts.interceptors) - - // Send messages immediately (multicast doesn't use lazy sending) - clientCtx.sendNow() - - // If waiting for send completion, drain the reply channel and return the first error. - if waitForSend { - var errs []nodeError - for range clientCtx.Size() { - select { - case r := <-clientCtx.replyChan: - if r.Err != nil && !errors.Is(r.Err, ErrSkipNode) { - errs = append(errs, nodeError{cause: r.Err, nodeID: r.NodeID}) - } - case <-ctx.Done(): - return ctx.Err() - } - } - if len(errs) > 0 { - return QuorumCallError{cause: ErrSendFailure, errors: errs} - } - } - return nil +// This function should be used by generated code only. +func Multicast[Req proto.Message](ctx *ConfigContext, req Req, method string) *OnewayCall[Req] { + return &OnewayCall[Req]{ctx: newOnewayCallContext(ctx, ctx.Config(), req, method)} } diff --git a/quorumcall.go b/quorumcall.go index 201a88ebd..e4bee46bb 100644 --- a/quorumcall.go +++ b/quorumcall.go @@ -1,55 +1,38 @@ package gorums -// QuorumCall performs a quorum call and returns a Responses object -// that provides access to node responses via terminal methods and fluent iteration. +import "google.golang.org/protobuf/proto" + +// QuorumCall performs a quorum call and returns a [Call] handle that provides +// access to node responses via terminal methods and fluent iteration. // // Type parameters: // - Req: The request message type // - Resp: The response message type from individual nodes // -// The opts parameter accepts CallOption values such as Interceptors. -// Interceptors are applied in the order they are provided via Interceptors -// while the client context is being constructed, before the user calls a terminal method. -// -// Note: Messages are not sent to nodes until a terminal method (like Majority, First) -// or iterator method (like Seq) is called, applying any registered request transformations. -// This lazy sending is necessary to allow interceptors to register transformations prior to dispatch. +// Register interceptors with [Call.Intercept] before invoking a terminal +// method. Messages are not sent to nodes until a terminal method (like Majority +// or First) or iterator method (like Results) is called, applying any registered +// request transformations. This lazy sending is what lets interceptors register +// transformations prior to dispatch. // // This function should only be used by generated code. -func QuorumCall[Req, Resp msg]( - ctx *ConfigContext, - req Req, - method string, - opts ...CallOption, -) *Responses[Resp] { - return invokeQuorumCall[Req, Resp](ctx, req, method, false, opts...) +func QuorumCall[Req, Resp proto.Message](ctx *ConfigContext, req Req, method string) *Call[Req, Resp] { + return invokeQuorumCall[Req, Resp](ctx, req, method, false) } -// QuorumCallStream performs a streaming quorum call and returns a Responses object. +// QuorumCallStream performs a streaming quorum call and returns a [Call] handle. // This is used for correctable stream methods where the server sends multiple responses. // // In streaming mode, the response iterator continues indefinitely until the context // is canceled, allowing the server to send multiple responses over time. // // This function should only be used by generated code. -func QuorumCallStream[Req, Resp msg]( - ctx *ConfigContext, - req Req, - method string, - opts ...CallOption, -) *Responses[Resp] { - return invokeQuorumCall[Req, Resp](ctx, req, method, true, opts...) +func QuorumCallStream[Req, Resp proto.Message](ctx *ConfigContext, req Req, method string) *Call[Req, Resp] { + return invokeQuorumCall[Req, Resp](ctx, req, method, true) } // invokeQuorumCall is the internal implementation shared by QuorumCall and QuorumCallStream. -func invokeQuorumCall[Req, Resp msg]( - ctx *ConfigContext, - req Req, - method string, - streaming bool, - opts ...CallOption, -) *Responses[Resp] { - callOpts := getCallOptions(opts...) - clientCtx := newQuorumCallContext[Req, Resp](ctx, req, method, streaming, callOpts.interceptors) - return NewResponses(clientCtx) +func invokeQuorumCall[Req, Resp proto.Message](ctx *ConfigContext, req Req, method string, streaming bool) *Call[Req, Resp] { + callCtx := newQuorumCallContext[Req, Resp](ctx, req, method, streaming) + return &Call[Req, Resp]{Responses: newResponses(callCtx), ctx: callCtx} } diff --git a/remote_call_test.go b/remote_call_test.go deleted file mode 100644 index 3c08bb94d..000000000 --- a/remote_call_test.go +++ /dev/null @@ -1,96 +0,0 @@ -package gorums_test - -import ( - "context" - "fmt" - "sync" - "testing" - "time" - - "github.com/relab/gorums" - "github.com/relab/gorums/gorumstest" - "github.com/relab/gorums/internal/testutils/mock" - pb "google.golang.org/protobuf/types/known/wrapperspb" -) - -func TestRPCCallSuccess(t *testing.T) { - node := gorumstest.Node(t, gorumstest.DefaultServer) - - ctx := gorumstest.Context(t, 5*time.Second) - nodeCtx := node.Context(ctx) - response, err := gorums.RemoteCall[*pb.StringValue, *pb.StringValue](nodeCtx, pb.String(""), mock.TestMethod) - if err != nil { - t.Fatalf("Unexpected error, got: %v, want: %v", err, nil) - } - if response == nil { - t.Fatalf("Unexpected response, got: %v, want: non-nil", nil) - } -} - -func TestRPCCallDownedNode(t *testing.T) { - node := gorumstest.Node(t, gorumstest.DefaultServer, gorumstest.WithPreConnect(t, func(stopServers func()) { - stopServers() - time.Sleep(300 * time.Millisecond) // wait for servers to fully stop - })) - - ctx := gorumstest.Context(t, 5*time.Second) - nodeCtx := node.Context(ctx) - response, err := gorums.RemoteCall[*pb.StringValue, *pb.StringValue](nodeCtx, pb.String(""), mock.TestMethod) - if err == nil { - t.Fatalf("Expected error, got: %v, want: %v", err, fmt.Errorf("rpc error: code = Unavailable desc = stream is down")) - } - if response != nil { - t.Fatalf("Unexpected response, got: %v, want: %v", response, nil) - } -} - -func TestRPCCallTimedOut(t *testing.T) { - node := gorumstest.Node(t, gorumstest.DefaultServer) - - ctx, cancel := context.WithTimeout(t.Context(), 0*time.Second) - time.Sleep(50 * time.Millisecond) - defer cancel() - nodeCtx := node.Context(ctx) - response, err := gorums.RemoteCall[*pb.StringValue, *pb.StringValue](nodeCtx, pb.String(""), mock.TestMethod) - if err == nil { - t.Fatalf("Expected error, got: %v, want: %v", err, fmt.Errorf("context deadline exceeded")) - } - if response != nil { - t.Fatalf("Unexpected response, got: %v, want: %v", response, nil) - } -} - -func TestRPCCallTypeMismatch(t *testing.T) { - node := gorumstest.Node(t, gorumstest.DefaultServer) - - ctx := gorumstest.Context(t, 5*time.Second) - nodeCtx := node.Context(ctx) - response, err := gorums.RemoteCall[*pb.StringValue, *pb.Int32Value](nodeCtx, pb.String(""), mock.TestMethod) - if err != gorums.ErrTypeMismatch { - t.Fatalf("Expected error, got: %v, want: %v", err, gorums.ErrTypeMismatch) - } - if response != nil { - t.Fatalf("Unexpected response, got: %v, want: %v", response, nil) - } -} - -func TestRPCCallConcurrentAccess(t *testing.T) { - node := gorumstest.Node(t, gorumstest.DefaultServer) - - concurrency := 10 - errCh := make(chan error, concurrency) - var wg sync.WaitGroup - for range concurrency { - wg.Go(func() { - _, err := gorums.RemoteCall[*pb.StringValue, *pb.StringValue](node.Context(t.Context()), pb.String(""), mock.TestMethod) - if err != nil { - errCh <- err - } - }) - } - wg.Wait() - close(errCh) - for err := range errCh { - t.Error(err) - } -} diff --git a/remote_call.go b/remotecall.go similarity index 79% rename from remote_call.go rename to remotecall.go index 573e2055c..496116c57 100644 --- a/remote_call.go +++ b/remotecall.go @@ -1,11 +1,14 @@ package gorums -import "github.com/relab/gorums/internal/stream" +import ( + "github.com/relab/gorums/internal/stream" + "google.golang.org/protobuf/proto" +) // RemoteCall executes a remote procedure call on the node. // // This method should be used by generated code only. -func RemoteCall[Req, Resp msg](ctx *NodeContext, req Req, method string) (Resp, error) { +func RemoteCall[Req, Resp proto.Message](ctx *NodeContext, req Req, method string) (Resp, error) { replyChan := make(chan NodeResponse[*stream.Message], 1) reqMsg, err := stream.NewMessage(ctx, ctx.nextMsgID(), method, req) if err != nil { diff --git a/responses.go b/responses.go index 111870068..d2d57ab17 100644 --- a/responses.go +++ b/responses.go @@ -9,16 +9,13 @@ import ( "github.com/relab/gorums/internal/stream" ) -// msg is a type alias for proto.Message intended to be used as a type parameter. -type msg = proto.Message - // NodeResponse is a type alias for stream.NodeResponse. type NodeResponse[T any] = stream.NodeResponse[T] // mapToCallResponse converts a NodeResponse[*stream.Message] to a NodeResponse[Resp]. // This is necessary because the channel layer's response router returns a // NodeResponse[*stream.Message] while the calltype expects a NodeResponse[Resp]. -func mapToCallResponse[Resp msg](channelResp NodeResponse[*stream.Message]) NodeResponse[Resp] { +func mapToCallResponse[Resp proto.Message](channelResp NodeResponse[*stream.Message]) NodeResponse[Resp] { callResp := NodeResponse[Resp]{ NodeID: channelResp.NodeID, Err: channelResp.Err, @@ -41,7 +38,7 @@ func mapToCallResponse[Resp msg](channelResp NodeResponse[*stream.Message]) Node // ------------------------------------------------------------------------- // ResponseSeq is an iterator that yields NodeResponse[T] values from a quorum call. -type ResponseSeq[T msg] iter.Seq[NodeResponse[T]] +type ResponseSeq[T proto.Message] iter.Seq[NodeResponse[T]] // IgnoreErrors returns an iterator that yields only successful responses, // discarding any responses with errors. This is useful when you want to process @@ -151,7 +148,7 @@ func (seq ResponseSeq[Resp]) CollectAll() map[uint32]Resp { // // Type parameter: // - Resp: The response message type -type Responses[Resp msg] struct { +type Responses[Resp proto.Message] struct { seq ResponseSeq[Resp] size int start starter @@ -159,9 +156,19 @@ type Responses[Resp msg] struct { type starter interface { sendNow() + markDispatched() +} + +// markDispatched marks the underlying call as dispatched without sending, so a +// later Intercept panics. Async and correctable calls use this before starting +// their goroutine. +func (r *Responses[Resp]) markDispatched() { + r.start.markDispatched() } -func NewResponses[Req, Resp msg](ctx *CallContext[Req, Resp]) *Responses[Resp] { +// newResponses builds the [Responses] handle returned by a quorum call from +// its [CallContext]. +func newResponses[Req, Resp proto.Message](ctx *CallContext[Req, Resp]) *Responses[Resp] { return &Responses[Resp]{ seq: ctx.responseSeq, size: ctx.Size(), @@ -176,7 +183,8 @@ func (r *Responses[Resp]) Size() int { // Results returns the underlying response iterator that yields node responses as they arrive. // It returns a single-use iterator. Users can use this to implement custom aggregation logic. -// This method triggers lazy sending of requests. +// This method triggers lazy sending of requests, and calling [Call.Intercept] +// after it panics. // // The iterator will: // - Yield responses as they arrive from nodes @@ -193,6 +201,7 @@ func (r *Responses[Resp]) Size() int { // // Process result.Value // } func (r *Responses[Resp]) Results() ResponseSeq[Resp] { + r.markDispatched() return r.seq } @@ -226,13 +235,18 @@ func (r *Responses[Resp]) All() (Resp, error) { // Threshold waits for a threshold number of successful responses. // It returns the first response once the threshold is reached. +// A response skipped by a request transform ([ErrSkipNode]) counts toward +// neither the successful-response count nor the reported node errors. func (r *Responses[Resp]) Threshold(threshold int) (resp Resp, err error) { var ( count int errs []nodeError ) for result := range r.seq { - if result.Err != nil && !errors.Is(result.Err, ErrSkipNode) { + if errors.Is(result.Err, ErrSkipNode) { + continue + } + if result.Err != nil { errs = append(errs, nodeError{nodeID: result.NodeID, cause: result.Err}) continue } diff --git a/responses_test.go b/responses_test.go index 595e4d238..3448f6396 100644 --- a/responses_test.go +++ b/responses_test.go @@ -6,12 +6,13 @@ import ( "github.com/relab/gorums/internal/stream" "github.com/relab/gorums/internal/testutils/mock" + "google.golang.org/protobuf/proto" pb "google.golang.org/protobuf/types/known/wrapperspb" ) // makeClientCtx is a helper to create a CallContext with mock responses for unit tests. // It creates a channel with the provided responses and returns a CallContext. -func makeClientCtx[Req, Resp msg](t *testing.T, numNodes int, responses []NodeResponse[msg]) *CallContext[Req, Resp] { +func makeClientCtx[Req, Resp proto.Message](t *testing.T, numNodes int, responses []NodeResponse[proto.Message]) *CallContext[Req, Resp] { t.Helper() resultChan := make(chan NodeResponse[*stream.Message], len(responses)) @@ -79,7 +80,7 @@ func TestTerminalMethods(t *testing.T) { tests := []struct { name string numNodes int - responses []NodeResponse[msg] + responses []NodeResponse[proto.Message] call func(resp respType) (*pb.StringValue, error) wantValue string wantErr bool @@ -89,7 +90,7 @@ func TestTerminalMethods(t *testing.T) { { name: "First_Success", numNodes: 3, - responses: []NodeResponse[msg]{ + responses: []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, }, call: respType.First, @@ -98,7 +99,7 @@ func TestTerminalMethods(t *testing.T) { { name: "First_Error", numNodes: 3, - responses: []NodeResponse[msg]{ + responses: []NodeResponse[proto.Message]{ {NodeID: 1, Value: nil, Err: errors.New("node error")}, {NodeID: 2, Value: nil, Err: errors.New("node error")}, {NodeID: 3, Value: nil, Err: errors.New("node error")}, @@ -111,7 +112,7 @@ func TestTerminalMethods(t *testing.T) { { name: "Majority_Success_3Nodes", numNodes: 3, - responses: []NodeResponse[msg]{ + responses: []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, {NodeID: 2, Value: pb.String("response2"), Err: nil}, }, @@ -121,7 +122,7 @@ func TestTerminalMethods(t *testing.T) { { name: "Majority_Insufficient", numNodes: 3, - responses: []NodeResponse[msg]{ + responses: []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, {NodeID: 2, Value: nil, Err: errors.New("node error")}, {NodeID: 3, Value: nil, Err: errors.New("node error")}, @@ -133,7 +134,7 @@ func TestTerminalMethods(t *testing.T) { { name: "Majority_Even_Success", numNodes: 4, - responses: []NodeResponse[msg]{ + responses: []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, {NodeID: 2, Value: pb.String("response2"), Err: nil}, {NodeID: 3, Value: pb.String("response3"), Err: nil}, @@ -145,7 +146,7 @@ func TestTerminalMethods(t *testing.T) { { name: "All_Success", numNodes: 3, - responses: []NodeResponse[msg]{ + responses: []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, {NodeID: 2, Value: pb.String("response2"), Err: nil}, {NodeID: 3, Value: pb.String("response3"), Err: nil}, @@ -156,7 +157,7 @@ func TestTerminalMethods(t *testing.T) { { name: "All_PartialFailure", numNodes: 3, - responses: []NodeResponse[msg]{ + responses: []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, {NodeID: 2, Value: pb.String("response2"), Err: nil}, {NodeID: 3, Value: nil, Err: errors.New("node error")}, @@ -169,7 +170,7 @@ func TestTerminalMethods(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { clientCtx := makeClientCtx[*pb.StringValue, *pb.StringValue](t, tt.numNodes, tt.responses) - responses := NewResponses(clientCtx) + responses := newResponses(clientCtx) result, err := tt.call(responses) @@ -188,7 +189,7 @@ func TestTerminalMethodsThreshold(t *testing.T) { tests := []struct { name string numNodes int - responses []NodeResponse[msg] + responses []NodeResponse[proto.Message] call func(resp respType, threshold int) (*pb.StringValue, error) threshold int wantValue string @@ -198,7 +199,7 @@ func TestTerminalMethodsThreshold(t *testing.T) { { name: "Threshold_Success", numNodes: 3, - responses: []NodeResponse[msg]{ + responses: []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, {NodeID: 2, Value: pb.String("response2"), Err: nil}, }, @@ -209,7 +210,7 @@ func TestTerminalMethodsThreshold(t *testing.T) { { name: "Threshold_Insufficient", numNodes: 3, - responses: []NodeResponse[msg]{ + responses: []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, {NodeID: 2, Value: nil, Err: errors.New("node error")}, {NodeID: 3, Value: nil, Err: errors.New("node error")}, @@ -223,7 +224,7 @@ func TestTerminalMethodsThreshold(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { clientCtx := makeClientCtx[*pb.StringValue, *pb.StringValue](t, tt.numNodes, tt.responses) - responses := NewResponses(clientCtx) + responses := newResponses(clientCtx) result, err := tt.call(responses, tt.threshold) @@ -244,13 +245,13 @@ func TestTerminalMethodsThreshold(t *testing.T) { // TestIteratorMethods tests the iterator helper methods func TestIteratorMethods(t *testing.T) { t.Run("IgnoreErrors", func(t *testing.T) { - responses := []NodeResponse[msg]{ + responses := []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, {NodeID: 2, Value: nil, Err: errors.New("node error")}, {NodeID: 3, Value: pb.String("response3"), Err: nil}, } clientCtx := makeClientCtx[*pb.StringValue, *pb.StringValue](t, 3, responses) - r := NewResponses(clientCtx) + r := newResponses(clientCtx) var count int for range r.Results().IgnoreErrors() { @@ -262,13 +263,13 @@ func TestIteratorMethods(t *testing.T) { }) t.Run("Filter", func(t *testing.T) { - responses := []NodeResponse[msg]{ + responses := []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, {NodeID: 2, Value: pb.String("response2"), Err: nil}, {NodeID: 3, Value: pb.String("response3"), Err: nil}, } clientCtx := makeClientCtx[*pb.StringValue, *pb.StringValue](t, 3, responses) - r := NewResponses(clientCtx) + r := newResponses(clientCtx) // Filter to only node 2 var count int @@ -286,13 +287,13 @@ func TestIteratorMethods(t *testing.T) { }) t.Run("CollectN", func(t *testing.T) { - responses := []NodeResponse[msg]{ + responses := []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, {NodeID: 2, Value: pb.String("response2"), Err: nil}, {NodeID: 3, Value: pb.String("response3"), Err: nil}, } clientCtx := makeClientCtx[*pb.StringValue, *pb.StringValue](t, 3, responses) - r := NewResponses(clientCtx) + r := newResponses(clientCtx) collected := r.Results().CollectN(2) if len(collected) != 2 { @@ -301,13 +302,13 @@ func TestIteratorMethods(t *testing.T) { }) t.Run("CollectAll", func(t *testing.T) { - responses := []NodeResponse[msg]{ + responses := []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, {NodeID: 2, Value: pb.String("response2"), Err: nil}, {NodeID: 3, Value: pb.String("response3"), Err: nil}, } clientCtx := makeClientCtx[*pb.StringValue, *pb.StringValue](t, 3, responses) - r := NewResponses(clientCtx) + r := newResponses(clientCtx) collected := r.Results().CollectAll() if len(collected) != 3 { @@ -336,13 +337,13 @@ func TestCustomAggregation(t *testing.T) { return nil, ErrIncomplete } - responses := []NodeResponse[msg]{ + responses := []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, {NodeID: 2, Value: pb.String("response2"), Err: nil}, {NodeID: 3, Value: pb.String("response3"), Err: nil}, } clientCtx := makeClientCtx[*pb.StringValue, *pb.StringValue](t, 3, responses) - r := NewResponses(clientCtx) + r := newResponses(clientCtx) // Call the aggregation function directly result, err := majorityQF(r) @@ -369,13 +370,13 @@ func TestCustomAggregation(t *testing.T) { return result, nil } - responses := []NodeResponse[msg]{ + responses := []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("alpha"), Err: nil}, {NodeID: 2, Value: pb.String("beta"), Err: nil}, {NodeID: 3, Value: pb.String("gamma"), Err: nil}, } clientCtx := makeClientCtx[*pb.StringValue, *pb.StringValue](t, 3, responses) - r := NewResponses(clientCtx) + r := newResponses(clientCtx) // Call the aggregation function directly - returns []string from *Responses[*pb.StringValue] result, err := collectAllValues(r) @@ -402,13 +403,13 @@ func TestCustomAggregation(t *testing.T) { return count, nil } - responses := []NodeResponse[msg]{ + responses := []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, {NodeID: 2, Value: pb.String("response2"), Err: nil}, {NodeID: 3, Value: pb.String("response3"), Err: nil}, } clientCtx := makeClientCtx[*pb.StringValue, *pb.StringValue](t, 3, responses) - r := NewResponses(clientCtx) + r := newResponses(clientCtx) // Call the aggregation function directly count, err := filterAndCount(r) @@ -438,12 +439,12 @@ func TestCustomAggregation(t *testing.T) { return first, nil } - responses := []NodeResponse[msg]{ + responses := []NodeResponse[proto.Message]{ {NodeID: 1, Value: pb.String("response1"), Err: nil}, {NodeID: 2, Value: nil, Err: errors.New("node 2 failed")}, } clientCtx := makeClientCtx[*pb.StringValue, *pb.StringValue](t, 2, responses) - r := NewResponses(clientCtx) + r := newResponses(clientCtx) // Call the aggregation function directly _, err := requireAllSuccess(r) diff --git a/server_e2e_test.go b/server_e2e_test.go index cb683c311..aeceb5acc 100644 --- a/server_e2e_test.go +++ b/server_e2e_test.go @@ -191,7 +191,7 @@ func outerChainedHandler( req, innerMethod, ) - res, err := respFn(responses) + res, err := respFn(responses.Responses) if err != nil { return nil, err } @@ -246,7 +246,7 @@ func TestServerSymmetricConfigurationRoutesQuorumCalls(t *testing.T) { mock.TestMethod, ) - result, err := tt.call(responses) + result, err := tt.call(responses.Responses) if err != nil { t.Fatalf("quorum call error: %v", err) } @@ -280,7 +280,7 @@ func TestServerSymmetricConfigurationRoutesMulticast(t *testing.T) { cfg.Context(ctx), pb.String("test"), mock.Stream, - ) + ).Send() if err != nil { t.Fatalf("multicast error: %v", err) } @@ -309,7 +309,7 @@ func TestServerHandlerCanMulticastViaConfig(t *testing.T) { cfg.Context(t.Context()), pb.String("inner-multicast"), mock.Stream, - ) + ).Send() if err != nil { return nil, err // failed to multicast } @@ -332,7 +332,7 @@ func TestServerHandlerCanMulticastViaConfig(t *testing.T) { cfg.Context(ctx), pb.String("outer-multicast"), mock.TestMethod, - ) + ).Send() if err != nil { t.Fatalf("multicast error: %v", err) } @@ -390,7 +390,7 @@ func TestServerHandlerCanChainQuorumCallViaConfig(t *testing.T) { pb.String("outer-call"), mock.TestMethod, ) - result, err := tt.outerFn(responses) + result, err := tt.outerFn(responses.Responses) if err != nil { t.Fatalf("quorum call error: %v", err) } @@ -449,7 +449,7 @@ func TestServerHandlerCanMulticastViaConnectedClients(t *testing.T) { cfg.Context(t.Context()), pb.String("inner-call"), mock.Stream, - ) + ).Send() if err != nil { return nil, err // failed to multicast } @@ -471,7 +471,7 @@ func TestServerHandlerCanMulticastViaConnectedClients(t *testing.T) { cfgClient.Context(ctx), pb.String("trigger"), mock.TestMethod, - ) + ).Send() if err != nil { t.Fatalf("multicast error: %v", err) } diff --git a/unicast.go b/unicast.go index e4bd69102..10fb6edfb 100644 --- a/unicast.go +++ b/unicast.go @@ -1,42 +1,20 @@ package gorums -import "github.com/relab/gorums/internal/stream" +import "google.golang.org/protobuf/proto" -// Unicast is a one-way call; no replies are returned to the client. +// Unicast is a one-way call to the single node in ctx; no reply is returned to +// the client. It returns an [OnewayCall] handle that dispatches the request only +// when it is consumed: [OnewayCall.Send] blocks until the send completes and +// reports any send error, while [OnewayCall.Async] dispatches without waiting +// and defers that error to [OnewayAsync.Wait]. // -// By default, this method blocks until the message has been sent to the node. -// This ensures that send operations complete before the caller proceeds, which can -// be useful for observing context cancellation or for pacing message sends. -// If the sending fails, the error is returned to the caller. +// Register request transforms with [OnewayCall.Intercept] and [MapRequest] +// before consuming the handle. // -// With the IgnoreErrors call option, the method returns nil immediately after -// enqueueing the message (fire-and-forget semantics). -// -// This method should be used by generated code only. -func Unicast[Req msg](ctx *NodeContext, req Req, method string, opts ...CallOption) error { - callOpts := getCallOptions(opts...) - reqMsg, err := stream.NewMessage(ctx, ctx.nextMsgID(), method, req) - if err != nil { - return err - } - - if callOpts.ignoreErrors { - // Fire-and-forget: enqueue and return immediately - ctx.enqueue(stream.Request{Ctx: ctx, Msg: reqMsg, Oneway: true}) - return nil - } - - // Default: block until send completes - replyChan := make(chan NodeResponse[*stream.Message], 1) - ctx.enqueue(stream.Request{Ctx: ctx, Msg: reqMsg, Oneway: true, ResponseChan: replyChan}) - - // Wait for send confirmation - select { - case r := <-replyChan: - // Unicast doesn't expect replies, but we still - // want to report errors from the send attempt. - return r.Err - case <-ctx.Done(): - return ctx.Err() +// This function should be used by generated code only. +func Unicast[Req proto.Message](ctx *NodeContext, req Req, method string) *OnewayCall[Req] { + return &OnewayCall[Req]{ + ctx: newOnewayCallContext(ctx, Config{ctx.Node()}, req, method), + unicast: true, } } From 7eb23f74c5c600ff128e725e0a0c004d0c96cd4c Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Tue, 11 Aug 2026 18:45:20 +0200 Subject: [PATCH 2/5] doc: update the user guide for the typed call handles Generated signatures no longer take call options, so the examples register interceptors with Intercept on the handle and consume one-way calls with Send or Async. Records that Intercept must precede any terminal method. Also repairs a set of section headings that a blanket Interceptor to ServerInterceptor rename caught by mistake: they describe client-side interceptors, not server-side ones. --- doc/user-guide.md | 94 ++++++++++++++++++++++------------------------- 1 file changed, 43 insertions(+), 51 deletions(-) diff --git a/doc/user-guide.md b/doc/user-guide.md index c1a588480..d8226b714 100644 --- a/doc/user-guide.md +++ b/doc/user-guide.md @@ -206,16 +206,17 @@ The first two functions are used to send requests to a single node determined by ```go func ReadRPC(ctx *gorums.NodeContext, in *ReadRequest) (resp *ReadResponse, err error) -func WriteUnicast(ctx *gorums.NodeContext, in *WriteRequest, opts ...gorums.CallOption) error +func WriteUnicast(ctx *gorums.NodeContext, in *WriteRequest) *gorums.OnewayCall[*WriteRequest] ``` The three functions below are used to send requests to a configuration of nodes determined by the `ConfigContext`. -The last two functions return a `*gorums.Responses[*ReadResponse]` object, which is a collection of responses from the nodes in the configuration. +A one-way call returns a `*gorums.OnewayCall` handle: call `Send()` to block until every send completes, or `Async()` to dispatch without waiting and `Wait()` for the result later. +The last two functions return a `*gorums.Call[*ReadRequest, *ReadResponse]` handle, from which the responses of the nodes in the configuration are aggregated. ```go -func WriteMulticast(ctx *gorums.ConfigContext, in *WriteRequest, opts ...gorums.CallOption) error -func ReadQC(ctx *gorums.ConfigContext, in *ReadRequest, opts ...gorums.CallOption) *gorums.Responses[*ReadResponse] -func ReadCorrectable(ctx *gorums.ConfigContext, in *ReadRequest, opts ...gorums.CallOption) *gorums.Responses[*ReadResponse] +func WriteMulticast(ctx *gorums.ConfigContext, in *WriteRequest) *gorums.OnewayCall[*WriteRequest] +func ReadQC(ctx *gorums.ConfigContext, in *ReadRequest) *gorums.Call[*ReadRequest, *ReadResponse] +func ReadCorrectable(ctx *gorums.ConfigContext, in *ReadRequest) *gorums.Call[*ReadRequest, *ReadResponse] ``` And this is our server interface: @@ -409,7 +410,7 @@ rpc ReadQC(ReadRequest) returns (ReadResponse) { The generated code provides a function for each quorum call method: ```go -func ReadQC(ctx *gorums.ConfigContext, in *ReadRequest, opts ...gorums.CallOption) *gorums.Responses[*ReadResponse] +func ReadQC(ctx *gorums.ConfigContext, in *ReadRequest) *gorums.Call[*ReadRequest, *ReadResponse] ``` This function returns a `*gorums.Responses[*ReadResponse]` object that provides several ways to aggregate and process responses. @@ -862,37 +863,34 @@ func RequireAllSuccess(resp *gorums.Responses[*Response]) (*Response, error) { ## Interceptors for Request/Response Transformation Gorums provides interceptors to transform requests and responses on a per-node basis. -Interceptors are passed as call options and can be chained together. +Register them with `Intercept` on the call handle, before any terminal method; they are applied in call-site order and may be chained. +Calling `Intercept` after the call has been dispatched panics, since an interceptor can no longer affect an in-flight call. -### MapRequest ClientInterceptor +### MapRequest Interceptor Transform requests before sending to each node: ```go cfgCtx := config.Context(ctx) -resp, err := WriteQC(cfgCtx, req, - gorums.Interceptors( - gorums.MapRequest(func(req *WriteRequest, node *gorums.Node) *WriteRequest { - // Customize request for each node - return &WriteRequest{Value: fmt.Sprintf("%s-node-%d", req.Value, node.ID())} - }), - ), +resp, err := WriteQC(cfgCtx, req).Intercept( + gorums.MapRequest(func(req *WriteRequest, node *gorums.Node) *WriteRequest { + // Customize request for each node + return &WriteRequest{Value: fmt.Sprintf("%s-node-%d", req.Value, node.ID())} + }), ).Majority() ``` -### MapResponse ClientInterceptor +### MapResponse Interceptor Transform responses received from each node: ```go -resp, err := ReadQC(cfgCtx, req, - gorums.Interceptors( - gorums.MapResponse(func(resp *ReadResponse, node *gorums.Node) *ReadResponse { - // Transform response, e.g., add node ID - resp.NodeID = node.ID() - return resp - }), - ), +resp, err := ReadQC(cfgCtx, req).Intercept( + gorums.MapResponse(func(resp *ReadResponse, node *gorums.Node) *ReadResponse { + // Transform response, e.g., add node ID + resp.NodeID = node.ID() + return resp + }), ).Majority() ``` @@ -901,13 +899,11 @@ resp, err := ReadQC(cfgCtx, req, ```go // Send different messages to each node in a multicast cfgCtx := config.Context(ctx) -WriteMulticast(cfgCtx, &WriteRequest{}, - gorums.Interceptors( - gorums.MapRequest(func(msg *WriteRequest, node *gorums.Node) *WriteRequest { - return &WriteRequest{Value: fmt.Sprintf("node-%d", node.ID())} - }), - ), -) +err := WriteMulticast(cfgCtx, &WriteRequest{}).Intercept( + gorums.MapRequest(func(msg *WriteRequest, node *gorums.Node) *WriteRequest { + return &WriteRequest{Value: fmt.Sprintf("node-%d", node.ID())} + }), +).Send() ``` **Note:** If `MapRequest` returns `nil` for a node, the message will not be sent to that node. @@ -937,20 +933,18 @@ The interceptor returns a new `ResponseSeq` that wraps `next` with custom logic. #### Chaining Interceptors -Multiple interceptors can be passed to `gorums.Interceptors()` and are executed in order: +Multiple interceptors can be passed to `Intercept` and are executed in order: ```go cfgCtx := config.Context(ctx) -resp, err := ReadQC(cfgCtx, req, - gorums.Interceptors( - loggingInterceptor, - gorums.MapRequest(transformFunc), - filterInterceptor, - ), +resp, err := ReadQC(cfgCtx, req).Intercept( + loggingInterceptor, + gorums.MapRequest(transformFunc), + filterInterceptor, ).Majority() ``` -#### Example: Logging ClientInterceptor +#### Example: Logging Interceptor Create a logging interceptor that wraps the response iterator: @@ -982,12 +976,12 @@ func LoggingInterceptor[Req, Resp proto.Message]( } // Usage -resp, err := ReadQC(cfgCtx, req, - gorums.Interceptors(LoggingInterceptor[*ReadRequest, *ReadResponse]), -).Majority() +resp, err := ReadQC(cfgCtx, req). + Intercept(LoggingInterceptor[*ReadRequest, *ReadResponse]). + Majority() ``` -#### Example: Response Filtering ClientInterceptor +#### Example: Response Filtering Interceptor Filter out responses that don't meet certain criteria: @@ -1012,16 +1006,14 @@ func FilterInterceptor[Req, Resp proto.Message]( // Usage: only include responses with timestamp > threshold threshold := time.Now().Add(-1 * time.Hour) -resp, err := ReadQC(cfgCtx, req, - gorums.Interceptors( - FilterInterceptor[*ReadRequest, *ReadResponse](func(r *ReadResponse) bool { - return r.GetTime().AsTime().After(threshold) - }), - ), +resp, err := ReadQC(cfgCtx, req).Intercept( + FilterInterceptor[*ReadRequest, *ReadResponse](func(r *ReadResponse) bool { + return r.GetTime().AsTime().After(threshold) + }), ).Majority() ``` -#### Example: Counting ClientInterceptor +#### Example: Counting Interceptor Count responses passing through the interceptor: @@ -1057,7 +1049,7 @@ You can pass multiple interceptors when starting a Gorums server. They can perfo Below are several examples based on the `examples/interceptors` package. -#### Server-Side Logging ServerInterceptor +#### Server-Side Logging Interceptor ```go func LoggingInterceptor(addr string) gorums.ServerInterceptor { From 4bed6cd839ca07efbeaa460c91edacf57f34e015 Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Tue, 11 Aug 2026 18:45:20 +0200 Subject: [PATCH 3/5] gorums: regenerate for the typed call handles Generated output only, produced by make genproto. Quorum calls return a Call handle, one-way calls return a OnewayCall handle, and neither takes call options. The one-way doc comments now show both terminals. --- .../dev/zorums_multicast_gorums.pb.go | 36 ++++++++++++++----- .../dev/zorums_quorumcall_gorums.pb.go | 18 ++++------ .../dev/zorums_unicast_gorums.pb.go | 30 +++++++++++----- examples/storage/proto/storage_gorums.pb.go | 33 +++++++++-------- internal/tests/config/config_gorums.pb.go | 3 +- .../correctable/correctable_gorums.pb.go | 6 ++-- internal/tests/oneway/oneway_gorums.pb.go | 28 +++++++++++---- internal/tests/ordering/order_gorums.pb.go | 3 +- 8 files changed, 100 insertions(+), 57 deletions(-) diff --git a/cmd/protoc-gen-gorums/dev/zorums_multicast_gorums.pb.go b/cmd/protoc-gen-gorums/dev/zorums_multicast_gorums.pb.go index 7ee52b737..fa0adbdd3 100644 --- a/cmd/protoc-gen-gorums/dev/zorums_multicast_gorums.pb.go +++ b/cmd/protoc-gen-gorums/dev/zorums_multicast_gorums.pb.go @@ -22,21 +22,41 @@ const ( var _ emptypb.Empty // Multicast plain. Response type is not needed here. -func Multicast(ctx *ConfigContext, in *Request, opts ...gorums.CallOption) error { - return gorums.Multicast(ctx, in, "dev.ZorumsService.Multicast", opts...) +// +// Example: +// +// err := Multicast(ctx, in).Send() +// h := Multicast(ctx, in).Async(); err := h.Wait() +func Multicast(ctx *ConfigContext, in *Request) *gorums.OnewayCall[*Request] { + return gorums.Multicast(ctx, in, "dev.ZorumsService.Multicast") } // Multicast2 is testing whether multiple streams work. -func Multicast2(ctx *ConfigContext, in *Request, opts ...gorums.CallOption) error { - return gorums.Multicast(ctx, in, "dev.ZorumsService.Multicast2", opts...) +// +// Example: +// +// err := Multicast2(ctx, in).Send() +// h := Multicast2(ctx, in).Async(); err := h.Wait() +func Multicast2(ctx *ConfigContext, in *Request) *gorums.OnewayCall[*Request] { + return gorums.Multicast(ctx, in, "dev.ZorumsService.Multicast2") } // Multicast3 is testing imported message type. -func Multicast3(ctx *ConfigContext, in *Request, opts ...gorums.CallOption) error { - return gorums.Multicast(ctx, in, "dev.ZorumsService.Multicast3", opts...) +// +// Example: +// +// err := Multicast3(ctx, in).Send() +// h := Multicast3(ctx, in).Async(); err := h.Wait() +func Multicast3(ctx *ConfigContext, in *Request) *gorums.OnewayCall[*Request] { + return gorums.Multicast(ctx, in, "dev.ZorumsService.Multicast3") } // Multicast4 is testing imported message type. -func Multicast4(ctx *ConfigContext, in *emptypb.Empty, opts ...gorums.CallOption) error { - return gorums.Multicast(ctx, in, "dev.ZorumsService.Multicast4", opts...) +// +// Example: +// +// err := Multicast4(ctx, in).Send() +// h := Multicast4(ctx, in).Async(); err := h.Wait() +func Multicast4(ctx *ConfigContext, in *emptypb.Empty) *gorums.OnewayCall[*emptypb.Empty] { + return gorums.Multicast(ctx, in, "dev.ZorumsService.Multicast4") } diff --git a/cmd/protoc-gen-gorums/dev/zorums_quorumcall_gorums.pb.go b/cmd/protoc-gen-gorums/dev/zorums_quorumcall_gorums.pb.go index 6641c24b7..5527c2287 100644 --- a/cmd/protoc-gen-gorums/dev/zorums_quorumcall_gorums.pb.go +++ b/cmd/protoc-gen-gorums/dev/zorums_quorumcall_gorums.pb.go @@ -19,50 +19,44 @@ const ( ) // QuorumCall plain. -func QuorumCall(ctx *ConfigContext, in *Request, opts ...gorums.CallOption) *gorums.Responses[*Response] { +func QuorumCall(ctx *ConfigContext, in *Request) *gorums.Call[*Request, *Response] { return gorums.QuorumCall[*Request, *Response]( ctx, in, "dev.ZorumsService.QuorumCall", - opts..., ) } // QuorumCallEmpty for testing imported message type. -func QuorumCallEmpty(ctx *ConfigContext, in *emptypb.Empty, opts ...gorums.CallOption) *gorums.Responses[*Response] { +func QuorumCallEmpty(ctx *ConfigContext, in *emptypb.Empty) *gorums.Call[*emptypb.Empty, *Response] { return gorums.QuorumCall[*emptypb.Empty, *Response]( ctx, in, "dev.ZorumsService.QuorumCallEmpty", - opts..., ) } // QuorumCallEmpty2 for testing imported message type. -func QuorumCallEmpty2(ctx *ConfigContext, in *Request, opts ...gorums.CallOption) *gorums.Responses[*emptypb.Empty] { +func QuorumCallEmpty2(ctx *ConfigContext, in *Request) *gorums.Call[*Request, *emptypb.Empty] { return gorums.QuorumCall[*Request, *emptypb.Empty]( ctx, in, "dev.ZorumsService.QuorumCallEmpty2", - opts..., ) } // QuorumCallStream plain. -func QuorumCallStream(ctx *ConfigContext, in *Request, opts ...gorums.CallOption) *gorums.Responses[*Response] { +func QuorumCallStream(ctx *ConfigContext, in *Request) *gorums.Call[*Request, *Response] { return gorums.QuorumCallStream[*Request, *Response]( ctx, in, "dev.ZorumsService.QuorumCallStream", - opts..., ) } // QuorumCallStreamWithEmpty for testing imported message type. -func QuorumCallStreamWithEmpty(ctx *ConfigContext, in *Request, opts ...gorums.CallOption) *gorums.Responses[*emptypb.Empty] { +func QuorumCallStreamWithEmpty(ctx *ConfigContext, in *Request) *gorums.Call[*Request, *emptypb.Empty] { return gorums.QuorumCallStream[*Request, *emptypb.Empty]( ctx, in, "dev.ZorumsService.QuorumCallStreamWithEmpty", - opts..., ) } // QuorumCallStreamWithEmpty2 for testing imported message type; with same return // type as QuorumCallStream: Response. -func QuorumCallStreamWithEmpty2(ctx *ConfigContext, in *emptypb.Empty, opts ...gorums.CallOption) *gorums.Responses[*Response] { +func QuorumCallStreamWithEmpty2(ctx *ConfigContext, in *emptypb.Empty) *gorums.Call[*emptypb.Empty, *Response] { return gorums.QuorumCallStream[*emptypb.Empty, *Response]( ctx, in, "dev.ZorumsService.QuorumCallStreamWithEmpty2", - opts..., ) } diff --git a/cmd/protoc-gen-gorums/dev/zorums_unicast_gorums.pb.go b/cmd/protoc-gen-gorums/dev/zorums_unicast_gorums.pb.go index 64d529abe..b140041ca 100644 --- a/cmd/protoc-gen-gorums/dev/zorums_unicast_gorums.pb.go +++ b/cmd/protoc-gen-gorums/dev/zorums_unicast_gorums.pb.go @@ -21,14 +21,28 @@ const ( // Reference imports to suppress errors if they are not otherwise used. var _ emptypb.Empty -// Unicast is a unicast call invoked on the node in ctx. -// No reply is returned to the client. -func Unicast(ctx *NodeContext, in *Request, opts ...gorums.CallOption) error { - return gorums.Unicast(ctx, in, "dev.ZorumsService.Unicast", opts...) +// Unicast is a unicast call invoked on the node in ctx; no reply is +// returned to the client. It returns a one-way call handle; call Send to block +// until the send completes and observe any send error, or Async to dispatch +// without waiting. +// +// Example: +// +// err := Unicast(ctx, in).Send() +// h := Unicast(ctx, in).Async(); err := h.Wait() +func Unicast(ctx *NodeContext, in *Request) *gorums.OnewayCall[*Request] { + return gorums.Unicast(ctx, in, "dev.ZorumsService.Unicast") } -// Unicast2 is a unicast call invoked on the node in ctx. -// No reply is returned to the client. -func Unicast2(ctx *NodeContext, in *Request, opts ...gorums.CallOption) error { - return gorums.Unicast(ctx, in, "dev.ZorumsService.Unicast2", opts...) +// Unicast2 is a unicast call invoked on the node in ctx; no reply is +// returned to the client. It returns a one-way call handle; call Send to block +// until the send completes and observe any send error, or Async to dispatch +// without waiting. +// +// Example: +// +// err := Unicast2(ctx, in).Send() +// h := Unicast2(ctx, in).Async(); err := h.Wait() +func Unicast2(ctx *NodeContext, in *Request) *gorums.OnewayCall[*Request] { + return gorums.Unicast(ctx, in, "dev.ZorumsService.Unicast2") } diff --git a/examples/storage/proto/storage_gorums.pb.go b/examples/storage/proto/storage_gorums.pb.go index 44145625c..8330c7a6d 100644 --- a/examples/storage/proto/storage_gorums.pb.go +++ b/examples/storage/proto/storage_gorums.pb.go @@ -67,58 +67,63 @@ func WriteRPC(ctx *NodeContext, in *WriteRequest) (*WriteResponse, error) { // WriteUnicast executes a one-way Write unicast call on a single node. // It does not wait for a response. -func WriteUnicast(ctx *NodeContext, in *WriteRequest, opts ...gorums.CallOption) error { - return gorums.Unicast(ctx, in, "proto.Storage.WriteUnicast", opts...) +// +// Example: +// +// err := WriteUnicast(ctx, in).Send() +// h := WriteUnicast(ctx, in).Async(); err := h.Wait() +func WriteUnicast(ctx *NodeContext, in *WriteRequest) *gorums.OnewayCall[*WriteRequest] { + return gorums.Unicast(ctx, in, "proto.Storage.WriteUnicast") } // WriteMulticast executes a Write multicast call on a configuration of nodes. // It does not wait for any responses. -func WriteMulticast(ctx *ConfigContext, in *WriteRequest, opts ...gorums.CallOption) error { - return gorums.Multicast(ctx, in, "proto.Storage.WriteMulticast", opts...) +// +// Example: +// +// err := WriteMulticast(ctx, in).Send() +// h := WriteMulticast(ctx, in).Async(); err := h.Wait() +func WriteMulticast(ctx *ConfigContext, in *WriteRequest) *gorums.OnewayCall[*WriteRequest] { + return gorums.Multicast(ctx, in, "proto.Storage.WriteMulticast") } // ReadQC executes a Read quorum call on a configuration of nodes and // returns the most recent value. -func ReadQC(ctx *ConfigContext, in *ReadRequest, opts ...gorums.CallOption) *gorums.Responses[*ReadResponse] { +func ReadQC(ctx *ConfigContext, in *ReadRequest) *gorums.Call[*ReadRequest, *ReadResponse] { return gorums.QuorumCall[*ReadRequest, *ReadResponse]( ctx, in, "proto.Storage.ReadQC", - opts..., ) } // WriteQC executes a Write quorum call on a configuration of nodes and // returns true if a majority of nodes were updated. -func WriteQC(ctx *ConfigContext, in *WriteRequest, opts ...gorums.CallOption) *gorums.Responses[*WriteResponse] { +func WriteQC(ctx *ConfigContext, in *WriteRequest) *gorums.Call[*WriteRequest, *WriteResponse] { return gorums.QuorumCall[*WriteRequest, *WriteResponse]( ctx, in, "proto.Storage.WriteQC", - opts..., ) } // ReadNestedQC executes a quorum call where each server handler performs // a nested quorum call using ServerCtx.Config(). -func ReadNestedQC(ctx *ConfigContext, in *ReadRequest, opts ...gorums.CallOption) *gorums.Responses[*ReadResponse] { +func ReadNestedQC(ctx *ConfigContext, in *ReadRequest) *gorums.Call[*ReadRequest, *ReadResponse] { return gorums.QuorumCall[*ReadRequest, *ReadResponse]( ctx, in, "proto.Storage.ReadNestedQC", - opts..., ) } // WriteNestedMulticast executes a quorum call where each server handler // performs a nested multicast using ServerCtx.Config(). -func WriteNestedMulticast(ctx *ConfigContext, in *WriteRequest, opts ...gorums.CallOption) *gorums.Responses[*WriteResponse] { +func WriteNestedMulticast(ctx *ConfigContext, in *WriteRequest) *gorums.Call[*WriteRequest, *WriteResponse] { return gorums.QuorumCall[*WriteRequest, *WriteResponse]( ctx, in, "proto.Storage.WriteNestedMulticast", - opts..., ) } // ReadCorrectable executes a quorum call that supports correctable responses. // It returns a stream of ReadResponse as multiple nodes respond or updates occur. -func ReadCorrectable(ctx *ConfigContext, in *ReadRequest, opts ...gorums.CallOption) *gorums.Responses[*ReadResponse] { +func ReadCorrectable(ctx *ConfigContext, in *ReadRequest) *gorums.Call[*ReadRequest, *ReadResponse] { return gorums.QuorumCallStream[*ReadRequest, *ReadResponse]( ctx, in, "proto.Storage.ReadCorrectable", - opts..., ) } diff --git a/internal/tests/config/config_gorums.pb.go b/internal/tests/config/config_gorums.pb.go index ac5d039fb..b89673ace 100644 --- a/internal/tests/config/config_gorums.pb.go +++ b/internal/tests/config/config_gorums.pb.go @@ -50,10 +50,9 @@ type CorrectableResponse = *gorums.Correctable[*Response] // Example: // // resp, err := Read(ctx, in).Majority() -func Read(ctx *ConfigContext, in *Request, opts ...gorums.CallOption) *gorums.Responses[*Response] { +func Read(ctx *ConfigContext, in *Request) *gorums.Call[*Request, *Response] { return gorums.QuorumCall[*Request, *Response]( ctx, in, "config.ConfigTest.Read", - opts..., ) } diff --git a/internal/tests/correctable/correctable_gorums.pb.go b/internal/tests/correctable/correctable_gorums.pb.go index 266bfdee9..1b5dec115 100644 --- a/internal/tests/correctable/correctable_gorums.pb.go +++ b/internal/tests/correctable/correctable_gorums.pb.go @@ -50,10 +50,9 @@ type CorrectableResponse = *gorums.Correctable[*Response] // Example: // // resp, err := Correctable(ctx, in).Majority() -func Correctable(ctx *ConfigContext, in *Request, opts ...gorums.CallOption) *gorums.Responses[*Response] { +func Correctable(ctx *ConfigContext, in *Request) *gorums.Call[*Request, *Response] { return gorums.QuorumCall[*Request, *Response]( ctx, in, "correctable.CorrectableTest.Correctable", - opts..., ) } @@ -65,10 +64,9 @@ func Correctable(ctx *ConfigContext, in *Request, opts ...gorums.CallOption) *go // corr := CorrectableStream(ctx, in).Correctable(2) // <-corr.Watch(2) // resp, level, err := corr.Get() -func CorrectableStream(ctx *ConfigContext, in *Request, opts ...gorums.CallOption) *gorums.Responses[*Response] { +func CorrectableStream(ctx *ConfigContext, in *Request) *gorums.Call[*Request, *Response] { return gorums.QuorumCallStream[*Request, *Response]( ctx, in, "correctable.CorrectableTest.CorrectableStream", - opts..., ) } diff --git a/internal/tests/oneway/oneway_gorums.pb.go b/internal/tests/oneway/oneway_gorums.pb.go index 5e7c76cac..c2e634ec9 100644 --- a/internal/tests/oneway/oneway_gorums.pb.go +++ b/internal/tests/oneway/oneway_gorums.pb.go @@ -37,16 +37,30 @@ type ( ConfigContext = gorums.ConfigContext ) -// Unicast is a unicast call invoked on the node in ctx. -// No reply is returned to the client. -func Unicast(ctx *NodeContext, in *Request, opts ...gorums.CallOption) error { - return gorums.Unicast(ctx, in, "oneway.OnewayTest.Unicast", opts...) +// Unicast is a unicast call invoked on the node in ctx; no reply is +// returned to the client. It returns a one-way call handle; call Send to block +// until the send completes and observe any send error, or Async to dispatch +// without waiting. +// +// Example: +// +// err := Unicast(ctx, in).Send() +// h := Unicast(ctx, in).Async(); err := h.Wait() +func Unicast(ctx *NodeContext, in *Request) *gorums.OnewayCall[*Request] { + return gorums.Unicast(ctx, in, "oneway.OnewayTest.Unicast") } // Multicast is a multicast call invoked on all nodes in the configuration in ctx. -// Use gorums.MapRequest to send different messages to each node. No replies are collected. -func Multicast(ctx *ConfigContext, in *Request, opts ...gorums.CallOption) error { - return gorums.Multicast(ctx, in, "oneway.OnewayTest.Multicast", opts...) +// It returns a one-way call handle; call Send to block until every send +// completes and observe any send errors, or Async to dispatch without waiting. +// Use gorums.MapRequest to send different messages to each node. +// +// Example: +// +// err := Multicast(ctx, in).Send() +// h := Multicast(ctx, in).Async(); err := h.Wait() +func Multicast(ctx *ConfigContext, in *Request) *gorums.OnewayCall[*Request] { + return gorums.Multicast(ctx, in, "oneway.OnewayTest.Multicast") } // OnewayTest is the server-side API for the OnewayTest Service diff --git a/internal/tests/ordering/order_gorums.pb.go b/internal/tests/ordering/order_gorums.pb.go index 7404af375..cdc9785f9 100644 --- a/internal/tests/ordering/order_gorums.pb.go +++ b/internal/tests/ordering/order_gorums.pb.go @@ -50,10 +50,9 @@ type CorrectableResponse = *gorums.Correctable[*Response] // Example: // // resp, err := QuorumCall(ctx, in).Majority() -func QuorumCall(ctx *ConfigContext, in *Request, opts ...gorums.CallOption) *gorums.Responses[*Response] { +func QuorumCall(ctx *ConfigContext, in *Request) *gorums.Call[*Request, *Response] { return gorums.QuorumCall[*Request, *Response]( ctx, in, "ordering.GorumsTest.QuorumCall", - opts..., ) } From ee8851437dcc2fbb621368d98864c51f18f5defd Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Wed, 12 Aug 2026 14:21:39 +0200 Subject: [PATCH 4/5] internal/tests/oneway: drain the received channels until empty reset evaluated len(srv.received) once and consumed exactly that many messages, so a straggler that arrived while it was draining stayed queued and was counted against the next subtest. Drain until each channel reports empty instead. --- internal/tests/oneway/oneway_test.go | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/internal/tests/oneway/oneway_test.go b/internal/tests/oneway/oneway_test.go index ab05a62fe..528bd45b1 100644 --- a/internal/tests/oneway/oneway_test.go +++ b/internal/tests/oneway/oneway_test.go @@ -52,10 +52,19 @@ type cluster struct { // reset discards messages left over from an earlier subtest so the next one // starts from a known state. A subtest that received everything it sent leaves // nothing behind. +// +// It drains until each channel is empty rather than taking len() once: a +// straggler still in flight when reset runs would otherwise be left queued and +// counted against the next subtest. func (c *cluster) reset() { for _, srv := range c.srvs { - for range len(srv.received) { - <-srv.received + drain: + for { + select { + case <-srv.received: + default: + break drain + } } } } From ba1465ab6471c107ecdc59e619510544e01c7337 Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Wed, 12 Aug 2026 14:21:39 +0200 Subject: [PATCH 5/5] doc: name the handle the quorum call helper returns The signature above it was updated to return *gorums.Call, but the prose still described a *gorums.Responses. Name the handle and say that it embeds Responses, which is what keeps the aggregation methods described below available on it. --- doc/user-guide.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/doc/user-guide.md b/doc/user-guide.md index d8226b714..aa8037a3b 100644 --- a/doc/user-guide.md +++ b/doc/user-guide.md @@ -413,7 +413,8 @@ The generated code provides a function for each quorum call method: func ReadQC(ctx *gorums.ConfigContext, in *ReadRequest) *gorums.Call[*ReadRequest, *ReadResponse] ``` -This function returns a `*gorums.Responses[*ReadResponse]` object that provides several ways to aggregate and process responses. +This function returns a `*gorums.Call[*ReadRequest, *ReadResponse]` handle. +The handle embeds `*gorums.Responses[*ReadResponse]`, so it offers the same ways to aggregate and process responses, and adds `Intercept` for registering interceptors before the call is dispatched. ### Terminal Methods for Response Aggregation