gorums: replace call options with typed call handles

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
+}

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)

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 {