fix(broadcast): client sends cancellation upon timeouts

Author: aleksander-vedvik

broadcast/shard_test.go

@@ -61,6 +61,7 @@ func TestShard(t *testing.T) {
 				OriginMethod:      "testMethod",
 				IsBroadcastClient: true,
 				ReceiveChan:       make(chan shardResponse),
+				Ctx:               context.Background(),
 			},
 			out: nil,
 		},
@@ -71,6 +72,7 @@ func TestShard(t *testing.T) {
 				OriginMethod:      "testMethod",
 				IsBroadcastClient: false,
 				ReceiveChan:       make(chan shardResponse),
+				Ctx:               context.Background(),
 			},
 			out: nil,
 		},
@@ -81,6 +83,7 @@ func TestShard(t *testing.T) {
 				OriginMethod:      "testMethod",
 				IsBroadcastClient: false,
 				ReceiveChan:       make(chan shardResponse),
+				Ctx:               context.Background(),
 			},
 			out: nil,
 		},

broadcastCall.go broadcastcall.go

@@ -7,10 +7,10 @@ import (
 	"google.golang.org/protobuf/reflect/protoreflect"
 )
 
-// broadcastCallData holds the message, destination nodes, method identifier,
+// BroadcastCallData holds the message, destination nodes, method identifier,
 // and other information necessary to perform the various quorum call types
 // supported by Gorums.
-type broadcastCallData struct {
+type BroadcastCallData struct {
 	Message           protoreflect.ProtoMessage
 	Method            string
 	BroadcastID       uint64 // a unique identifier for the current broadcast request
@@ -23,7 +23,7 @@ type broadcastCallData struct {
 
 // checks whether the given address is contained in the given subset
 // of server addresses. Will return true if a subset is not given.
-func (bcd *broadcastCallData) inSubset(addr string) bool {
+func (bcd *BroadcastCallData) inSubset(addr string) bool {
 	if bcd.ServerAddresses == nil || len(bcd.ServerAddresses) <= 0 {
 		return true
 	}
@@ -35,8 +35,8 @@ func (bcd *broadcastCallData) inSubset(addr string) bool {
 	return false
 }
 
-// broadcastCall performs a multicast call on the configuration.
-func (c RawConfiguration) broadcastCall(ctx context.Context, d broadcastCallData) {
+// BroadcastCall performs a multicast call on the configuration.
+func (c RawConfiguration) BroadcastCall(ctx context.Context, d BroadcastCallData) {
 	md := &ordering.Metadata{MessageID: c.getMsgID(), Method: d.Method, BroadcastMsg: &ordering.BroadcastMsg{
 		IsBroadcastClient: d.IsBroadcastClient,
 		BroadcastID:       d.BroadcastID,

clientserver.go

@@ -99,7 +99,6 @@ func createReq(ctx, clientCtx context.Context, cancel context.CancelFunc, req pr
 		select {
 		case <-clientCtx.Done():
 			// client provided ctx
-			//slog.Info("clientserver: clientCtx done", "resps", len(resps))
 			return
 		case <-ctx.Done():
 			// request ctx. this is a child to the server ctx.

cmd/protoc-gen-gorums/dev/zorums_broadcastcall_gorums.pb.go

@@ -23,13 +23,19 @@ func (c *Configuration) {{.Method.GoName}}(ctx context.Context, in *{{in .GenFil
 	if c.qspec == nil {
 		return nil, fmt.Errorf("a qspec is not defined")
 	}
-	doneChan, cd := c.srv.AddRequest(c.snowflake.NewBroadcastID(), ctx, in, gorums.ConvertToType(c.qspec.{{.Method.GoName}}QF), "{{.Method.Desc.FullName}}")
+	broadcastID := c.snowflake.NewBroadcastID()
+	doneChan, cd := c.srv.AddRequest(broadcastID, ctx, in, gorums.ConvertToType(c.qspec.{{.Method.GoName}}QF), "{{.Method.Desc.FullName}}")
 	c.RawConfiguration.Multicast(ctx, cd, gorums.WithNoSendWaiting())
 	var response {{$protoMessage}}
 	var ok bool
 	select {
 	case response, ok = <-doneChan:
 	case <-ctx.Done():
+		bd := gorums.BroadcastCallData{
+			Method:      gorums.Cancellation,
+			BroadcastID: broadcastID,
+		}
+		c.RawConfiguration.BroadcastCall(context.Background(), bd)
 		return nil, fmt.Errorf("context cancelled")
 	}
 	if !ok {

cmd/protoc-gen-gorums/dev/zorums_qspec_gorums.pb.go

@@ -132,7 +132,7 @@ func (srv *broadcastServer) sendToClientHandler(broadcastID uint64, resp protore
 }
 
 func (srv *broadcastServer) forwardHandler(req RequestTypes, method string, broadcastID uint64, forwardAddr, originAddr string) {
-	cd := broadcastCallData{
+	cd := BroadcastCallData{
 		Message:           req,
 		Method:            method,
 		BroadcastID:       broadcastID,
@@ -142,7 +142,7 @@ func (srv *broadcastServer) forwardHandler(req RequestTypes, method string, broa
 	}
 	srv.viewMutex.RLock()
 	// drop request if a view change has occured
-	srv.view.broadcastCall(context.Background(), cd)
+	srv.view.BroadcastCall(context.Background(), cd)
 	srv.viewMutex.RUnlock()
 }
 
@@ -155,21 +155,20 @@ func (srv *broadcastServer) doneHandler(broadcastID uint64) {
 }
 
 func (srv *broadcastServer) canceler(broadcastID uint64, srvAddrs []string) {
-	cd := broadcastCallData{
-		//Message:         &emptypb.Empty{},
+	cd := BroadcastCallData{
 		Message:         nil,
 		Method:          Cancellation,
 		BroadcastID:     broadcastID,
 		ServerAddresses: srvAddrs,
 	}
 	srv.viewMutex.RLock()
 	// drop request if a view change has occured
-	srv.view.broadcastCall(context.Background(), cd)
+	srv.view.BroadcastCall(context.Background(), cd)
 	srv.viewMutex.RUnlock()
 }
 
 func (srv *broadcastServer) serverBroadcastHandler(method string, req RequestTypes, opts ...broadcast.BroadcastOptions) {
-	cd := broadcastCallData{
+	cd := BroadcastCallData{
 		Message:           req,
 		Method:            method,
 		BroadcastID:       srv.manager.NewBroadcastID(),
@@ -178,7 +177,7 @@ func (srv *broadcastServer) serverBroadcastHandler(method string, req RequestTyp
 	}
 	srv.viewMutex.RLock()
 	// drop request if a view change has occured
-	srv.view.broadcastCall(context.Background(), cd)
+	srv.view.BroadcastCall(context.Background(), cd)
 	srv.viewMutex.RUnlock()
 }
 
@@ -188,7 +187,7 @@ func (srv *Server) SendToClientHandler(resp protoreflect.ProtoMessage, err error
 
 func (srv *broadcastServer) registerBroadcastFunc(method string) {
 	srv.manager.AddHandler(method, broadcast.ServerHandler(func(ctx context.Context, in protoreflect.ProtoMessage, broadcastID uint64, originAddr, originMethod string, options broadcast.BroadcastOptions, id uint32, addr string) {
-		cd := broadcastCallData{
+		cd := BroadcastCallData{
 			Message:           in,
 			Method:            method,
 			BroadcastID:       broadcastID,
@@ -200,7 +199,7 @@ func (srv *broadcastServer) registerBroadcastFunc(method string) {
 		}
 		srv.viewMutex.RLock()
 		// drop request if a view change has occured
-		srv.view.broadcastCall(ctx, cd)
+		srv.view.BroadcastCall(ctx, cd)
 		srv.viewMutex.RUnlock()
 	}))
 	/*srv.manager.AddServerHandler(method, func(ctx context.Context, in protoreflect.ProtoMessage, broadcastID uint64, originAddr, originMethod string, options broadcast.BroadcastOptions, id uint32, addr string) {

cmd/protoc-gen-gorums/gengorums/template_broadcastcall.go

@@ -29,7 +29,7 @@ func (c RawConfiguration) Multicast(ctx context.Context, d QuorumCallData, opts
 				continue // don't send if no msg
 			}
 		}
-		go n.channel.enqueue(request{ctx: ctx, msg: &Message{Metadata: md, Message: msg}, opts: o}, replyChan, false)
+		n.channel.enqueue(request{ctx: ctx, msg: &Message{Metadata: md, Message: msg}, opts: o}, replyChan, false)
 		sentMsgs++
 	}
 

handler.go

@@ -68,9 +68,9 @@ func (n *RawNode) connect(mgr *RawManager) error {
 		return nil
 	}
 	n.channel = newChannel(n)
-	//if err := n.channel.connect(); err != nil {
-	//	return nodeError{nodeID: n.id, cause: err}
-	//}
+	if err := n.channel.connect(); err != nil {
+		return nodeError{nodeID: n.id, cause: err}
+	}
 	return nil
 }