feat(broadcast): added logging

Author: aleksander-vedvik

broadcast/processor.go

@@ -2,6 +2,7 @@ package broadcast
 
 import (
 	"context"
+	"log/slog"
 	"time"
 
 	"google.golang.org/protobuf/reflect/protoreflect"
@@ -16,6 +17,7 @@ type BroadcastProcessor struct {
 	cancelFunc    context.CancelFunc
 	started       time.Time
 	ended         time.Time
+	logger        *slog.Logger
 
 	cancellationCtx       context.Context
 	cancellationCtxCancel context.CancelFunc
@@ -60,12 +62,14 @@ func (p *BroadcastProcessor) handle(msg Content) {
 			msg.ReceiveChan <- shardResponse{
 				err: OutOfOrderErr{},
 			}
+			p.log("msg: out of order", "err", OutOfOrderErr{}, "method", msg.CurrentMethod, "from", msg.SenderAddr)
 		} else {
 			msg.ReceiveChan <- shardResponse{
 				err:              nil,
 				reqCtx:           p.cancellationCtx,
 				enqueueBroadcast: p.enqueueBroadcast,
 			}
+			p.log("msg: processed", "err", nil, "method", msg.CurrentMethod, "from", msg.SenderAddr)
 		}
 	}
 	defer func() {
@@ -79,13 +83,15 @@ func (p *BroadcastProcessor) handle(msg Content) {
 		//close(p.broadcastChan)
 		//close(p.sendChan)
 		p.emptyChannels(metadata)
+		p.log("processor stopped", "err", nil, "started", p.started, "ended", p.ended)
 	}()
 	for {
 		select {
 		case <-p.ctx.Done():
 			return
 		case bMsg := <-p.broadcastChan:
 			if p.broadcastID != bMsg.BroadcastID {
+				p.log("broadcast: wrong BroadcastID", "err", BroadcastIDErr{}, "type", bMsg.MsgType.String(), "stopping", false)
 				continue
 			}
 			switch bMsg.MsgType {
@@ -110,6 +116,7 @@ func (p *BroadcastProcessor) handle(msg Content) {
 				new.ReceiveChan <- shardResponse{
 					err: BroadcastIDErr{},
 				}
+				p.log("msg: wrong BroadcastID", "err", BroadcastIDErr{}, "method", new.CurrentMethod, "from", new.SenderAddr)
 				continue
 			}
 			if new.IsCancellation {
@@ -119,6 +126,7 @@ func (p *BroadcastProcessor) handle(msg Content) {
 				new.ReceiveChan <- shardResponse{
 					err: nil,
 				}
+				p.log("msg: received cancellation", "err", nil, "method", new.CurrentMethod, "from", new.SenderAddr)
 				continue
 			}
 
@@ -129,6 +137,7 @@ func (p *BroadcastProcessor) handle(msg Content) {
 					new.ReceiveChan <- shardResponse{
 						err: ClientReqAlreadyReceivedErr{},
 					}
+					p.log("msg: duplicate client req", "err", ClientReqAlreadyReceivedErr{}, "method", new.CurrentMethod, "from", new.SenderAddr)
 					continue
 				}
 				// important to set this option to prevent duplicate client reqs.
@@ -146,6 +155,7 @@ func (p *BroadcastProcessor) handle(msg Content) {
 					}
 					p.cancellationCtxCancel()
 				}()
+				p.log("msg: received client req", "err", nil, "method", new.CurrentMethod, "from", new.SenderAddr)
 			}
 
 			metadata.update(new)
@@ -164,6 +174,7 @@ func (p *BroadcastProcessor) handle(msg Content) {
 					err: err,
 				}
 				//	slog.Info("receive: late", "err", err, "id", p.broadcastID)
+				p.log("msg: late msg", "err", err, "method", new.CurrentMethod, "from", new.SenderAddr)
 				return
 			}
 			if !p.isInOrder(new.CurrentMethod) {
@@ -172,22 +183,26 @@ func (p *BroadcastProcessor) handle(msg Content) {
 				new.ReceiveChan <- shardResponse{
 					err: OutOfOrderErr{},
 				}
+				p.log("msg: out of order", "err", OutOfOrderErr{}, "method", new.CurrentMethod, "from", new.SenderAddr)
 				continue
 			}
 			new.ReceiveChan <- shardResponse{
 				err:              nil,
 				reqCtx:           p.cancellationCtx,
 				enqueueBroadcast: p.enqueueBroadcast,
 			}
+			p.log("msg: processed", "err", nil, "method", new.CurrentMethod, "from", new.SenderAddr)
 		}
 	}
 }
 
 func (p *BroadcastProcessor) handleCancellation(bMsg Msg, metadata *metadata) bool {
 	if bMsg.Cancellation.end {
+		p.log("broadcast: broadcast.Done() called", "err", nil, "type", bMsg.MsgType.String(), "stopping", true)
 		return true
 	}
 	if !metadata.SentCancellation {
+		p.log("broadcast: sent cancellation", "err", nil, "type", bMsg.MsgType.String(), "stopping", false)
 		metadata.SentCancellation = true
 		go p.router.Send(p.broadcastID, "", "", bMsg.Cancellation)
 	}
@@ -201,17 +216,25 @@ func (p *BroadcastProcessor) handleBroadcast(bMsg Msg, methods []string, metadat
 		return false
 	}
 	p.router.Send(p.broadcastID, metadata.OriginAddr, metadata.OriginMethod, bMsg.Msg)
+	p.log("broadcast: sending broadcast", "err", nil, "type", bMsg.MsgType.String(), "stopping", false, "isBroadcastCall", metadata.isBroadcastCall())
 
 	p.updateOrder(bMsg.Method)
 	p.dispatchOutOfOrderMsgs()
 	return true
 }
 
+func (p *BroadcastProcessor) log(msg string, args ...any) {
+	if p.logger != nil {
+		p.logger.Debug(msg, args...)
+	}
+}
+
 func (p *BroadcastProcessor) handleReply(bMsg Msg, metadata *metadata) bool {
 	// BroadcastCall if origin addr is non-empty.
 	if metadata.isBroadcastCall() {
 		go p.router.Send(p.broadcastID, metadata.OriginAddr, metadata.OriginMethod, bMsg.Reply)
 		// the request is done becuase we have sent a reply to the client
+		p.log("broadcast: sending reply to client", "err", nil, "type", bMsg.MsgType.String(), "stopping", true, "isBroadcastCall", metadata.isBroadcastCall())
 		return true
 	}
 	// QuorumCall if origin addr is empty.
@@ -231,9 +254,11 @@ func (p *BroadcastProcessor) handleReply(bMsg Msg, metadata *metadata) bool {
 		// the request is not done yet because we have not replied to
 		// the client.
 		//slog.Info("reply: late", "err", err, "id", p.broadcastID)
+		p.log("broadcast: failed to send reply to client", "err", err, "type", bMsg.MsgType.String(), "stopping", false, "isBroadcastCall", metadata.isBroadcastCall())
 		return false
 	}
 	// the request is done becuase we have sent a reply to the client
+	p.log("broadcast: sending reply to client", "err", err, "type", bMsg.MsgType.String(), "stopping", true, "isBroadcastCall", metadata.isBroadcastCall())
 	return true
 }
 
@@ -394,6 +419,7 @@ func (r *BroadcastProcessor) dispatchOutOfOrderMsgs() {
 		if order <= r.orderIndex {
 			for _, msg := range msgs {
 				msg.Run(r.cancellationCtx, r.enqueueBroadcast)
+				r.log("msg: dispatching out of order msg", "err", nil, "method", msg.CurrentMethod, "from", msg.SenderAddr)
 			}
 			handledMethods = append(handledMethods, method)
 		}

broadcast/router.go

@@ -70,7 +70,9 @@ func (r *BroadcastRouter) Send(broadcastID uint64, addr, method string, req any)
 		r.canceler(broadcastID, val.srvAddrs)
 		return nil
 	}
-	return errors.New("wrong req type")
+	err := errors.New("wrong req type")
+	r.log("router: malformed msg", "err", err, "BroadcastID", broadcastID)
+	return err
 }
 
 func (r *BroadcastRouter) Connect(addr string) {
@@ -84,21 +86,28 @@ func (r *BroadcastRouter) routeBroadcast(broadcastID uint64, addr, method string
 		handler(msg.ctx, msg.request, broadcastID, addr, method, msg.options, r.id, r.addr)
 		return nil
 	}
-	return errors.New("not found")
+	err := errors.New("handler not found")
+	r.log("router (broadcast): could not find handler", "err", err, "BroadcastID", broadcastID, "addr", addr, "method", method)
+	return err
 }
 
 func (r *BroadcastRouter) routeClientReply(broadcastID uint64, addr, method string, resp *reply) error {
 	// the client has initiated a broadcast call and the reply should be sent as an RPC
 	if _, ok := r.clientHandlers[method]; ok && addr != "" {
 		client, err := r.getClient(addr)
 		if err != nil {
+			r.log("router (reply): could not get client", "err", err, "BroadcastID", broadcastID, "addr", addr, "method", method)
 			return err
 		}
-		return client.SendMsg(broadcastID, method, resp.getResponse(), r.dialTimeout)
+		err = client.SendMsg(broadcastID, method, resp.getResponse(), r.dialTimeout)
+		r.log("router (reply): could not send reply", "err", err, "BroadcastID", broadcastID, "addr", addr, "method", method)
+		return err
 	}
 	// the server can receive a broadcast from another server before a client sends a direct message.
 	// it should thus wait for a potential message from the client. otherwise, it should be removed.
-	return errors.New("not routed")
+	err := errors.New("not routed")
+	r.log("router (reply): could not find handler", "err", err, "BroadcastID", broadcastID, "addr", addr, "method", method)
+	return err
 }
 
 func (r *BroadcastRouter) getClient(addr string) (*Client, error) {
@@ -132,6 +141,12 @@ func (r *BroadcastRouter) getClient(addr string) (*Client, error) {
 	return client, nil
 }
 
+func (r *BroadcastRouter) log(msg string, args ...any) {
+	if r.logger != nil {
+		r.logger.Debug(msg, args...)
+	}
+}
+
 type msgType int
 
 const (

broadcast/shard.go

@@ -2,6 +2,7 @@ package broadcast
 
 import (
 	"context"
+	"log/slog"
 	"sync"
 	"time"
 )
@@ -37,12 +38,13 @@ type shard struct {
 	nextGC      time.Time
 	shardBuffer int
 	sendBuffer  int
+	logger      *slog.Logger
 
 	preserveOrdering bool
 	order            map[string]int
 }
 
-func createShards(ctx context.Context, shardBuffer, sendBuffer int, router Router, order map[string]int, reqTTL time.Duration) []*shard {
+func createShards(ctx context.Context, shardBuffer, sendBuffer int, router Router, order map[string]int, reqTTL time.Duration, logger *slog.Logger) []*shard {
 	shards := make([]*shard, NumShards)
 	for i := range shards {
 		ctx, cancel := context.WithCancel(ctx)
@@ -60,6 +62,7 @@ func createShards(ctx context.Context, shardBuffer, sendBuffer int, router Route
 			router:           router,
 			preserveOrdering: order != nil,
 			order:            order,
+			logger:           logger,
 		}
 	}
 	return shards
@@ -375,6 +378,10 @@ func (s *shard) addProcessor2(sendBuffer int, msg Content) (*BroadcastProcessor,
 	// should only affect the current shard and not the others.
 	ctx, cancel := context.WithTimeout(s.ctx, s.reqTTL)
 	//req := &BroadcastRequest{
+	var logger *slog.Logger
+	if s.logger != nil {
+		logger = s.logger.With(slog.Uint64("BroadcastID", msg.BroadcastID))
+	}
 	req := &BroadcastProcessor{
 		ctx:        ctx,
 		cancelFunc: cancel,
@@ -395,6 +402,7 @@ func (s *shard) addProcessor2(sendBuffer int, msg Content) (*BroadcastProcessor,
 		cancellationCtxCancel: msg.CancelCtx,
 		executionOrder:        s.order,
 		//sendCancellation:      new(sync.Once),
+		logger: logger,
 	}
 	s.reqs[msg.BroadcastID] = req
 	//go req.handle(s.router, msg.BroadcastID, msg)

broadcast/state.go

@@ -55,7 +55,7 @@ func NewState(logger *slog.Logger, router Router, order map[string]int) *Broadca
 	sendBuffer := 30
 	TTL := 5 * time.Minute
 	ctx, cancel := context.WithCancel(context.Background())
-	shards := createShards(ctx, shardBuffer, sendBuffer, router, order, TTL)
+	shards := createShards(ctx, shardBuffer, sendBuffer, router, order, TTL, logger)
 	state := &BroadcastState{
 		parentCtx:           ctx,
 		parentCtxCancelFunc: cancel,
@@ -116,7 +116,7 @@ func (s *BroadcastState) reset() {
 	ctx, cancel := context.WithCancel(context.Background())
 	s.parentCtx = ctx
 	s.parentCtxCancelFunc = cancel
-	s.shards = createShards(ctx, s.shardBuffer, s.sendBuffer, s.router, s.order, s.reqTTL)
+	s.shards = createShards(ctx, s.shardBuffer, s.sendBuffer, s.router, s.order, s.reqTTL, s.logger)
 	s.RunShards()
 	for _, client := range s.clients {
 		client.Close()
@@ -173,6 +173,7 @@ type Content struct {
 	IsCancellation    bool
 	OriginAddr        string
 	OriginMethod      string
+	SenderAddr        string
 	CurrentMethod     string
 	ReceiveChan       chan shardResponse
 	SendFn            func(resp protoreflect.ProtoMessage, err error) error

broadcastcall.go

@@ -35,16 +35,16 @@ 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 broadcast on the configuration.
+func (c RawConfiguration) BroadcastCall(ctx context.Context, d BroadcastCallData, opts ...CallOption) {
 	md := &ordering.Metadata{MessageID: c.getMsgID(), Method: d.Method, BroadcastMsg: &ordering.BroadcastMsg{
 		IsBroadcastClient: d.IsBroadcastClient,
 		BroadcastID:       d.BroadcastID,
 		SenderAddr:        d.SenderAddr,
 		OriginAddr:        d.OriginAddr,
 		OriginMethod:      d.OriginMethod,
 	}}
-	o := getCallOptions(E_Broadcast, nil)
+	o := getCallOptions(E_Broadcast, opts)
 
 	var replyChan chan response
 	if !o.noSendWaiting {
@@ -82,6 +82,7 @@ func (c RawConfiguration) BroadcastCall(ctx context.Context, d BroadcastCallData
 		n.channel.enqueueSlow(request{ctx: ctx, msg: &Message{Metadata: md, Message: msg}, opts: o})
 	}
 
+	// if noSendWaiting is set, we will not wait for confirmation from the channel before returning.
 	if o.noSendWaiting {
 		return
 	}
@@ -90,5 +91,4 @@ func (c RawConfiguration) BroadcastCall(ctx context.Context, d BroadcastCallData
 	for ; sentMsgs > 0; sentMsgs-- {
 		<-replyChan
 	}
-	close(replyChan)
 }

clientserver.go

@@ -64,8 +64,8 @@ func (srv *ClientServer) Stop() {
 	}
 }
 
-func (srv *ClientServer) AddRequest(broadcastID uint64, clientCtx context.Context, in protoreflect.ProtoMessage, handler ReplySpecHandler, method string) (chan protoreflect.ProtoMessage, QuorumCallData) {
-	cd := QuorumCallData{
+func (srv *ClientServer) AddRequest(broadcastID uint64, clientCtx context.Context, in protoreflect.ProtoMessage, handler ReplySpecHandler, method string) (chan protoreflect.ProtoMessage, BroadcastCallData) {
+	cd := BroadcastCallData{
 		Message: in,
 		Method:  method,