diff --git a/call.go b/call.go new file mode 100644 index 00000000..51004c89 --- /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 1ae83699..01a28e1c 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 a9e47d43..42790e67 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 b20e6311..cacbdf21 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 16c8c6f4..852a762a 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 00000000..2a18801e --- /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 8488e9ad..00000000 --- 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 e7aa5e62..00000000 --- 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/dev/zorums_multicast_gorums.pb.go b/cmd/protoc-gen-gorums/dev/zorums_multicast_gorums.pb.go index 7ee52b73..fa0adbdd 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 6641c24b..5527c228 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 64d529ab..b140041c 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/cmd/protoc-gen-gorums/gengorums/template_multicast.go b/cmd/protoc-gen-gorums/gengorums/template_multicast.go index 055550c3..1dc6056c 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 6c767807..353e3a85 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 9eb0442a..af663032 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 58e5da87..25bc84d4 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/doc/user-guide.md b/doc/user-guide.md index c1a58848..aa8037a3 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,10 +410,11 @@ 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. +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 @@ -862,37 +864,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 +900,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 +934,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 +977,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 +1007,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 +1050,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 { diff --git a/examples/storage/client.go b/examples/storage/client.go index d8307d1b..85a5cf92 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/proto/storage_gorums.pb.go b/examples/storage/proto/storage_gorums.pb.go index 44145625..8330c7a6 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/examples/storage/repl.go b/examples/storage/repl.go index 39beac70..0d149913 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 acac58bb..802fad08 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 1c0bb039..95509c1d 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/config/config_gorums.pb.go b/internal/tests/config/config_gorums.pb.go index ac5d039f..b89673ac 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 266bfdee..1b5dec11 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 5e7c76ca..c2e634ec 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/oneway/oneway_test.go b/internal/tests/oneway/oneway_test.go index 529593f4..528bd45b 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,136 @@ 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. +// +// 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 { + drain: + for { + select { + case <-srv.received: + default: + break drain + } + } + } +} + +// 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 +203,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 +254,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 +269,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/internal/tests/ordering/order_gorums.pb.go b/internal/tests/ordering/order_gorums.pb.go index 7404af37..cdc9785f 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..., ) } diff --git a/multicast.go b/multicast.go index 9e12ed0a..6aa75000 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 201a88eb..e4bee46b 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 3c08bb94..00000000 --- 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 573e2055..496116c5 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 11187006..d2d57ab1 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 595e4d23..3448f639 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 cb683c31..aeceb5ac 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 e4bd6910..10fb6edf 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, } }