fix(broadcast): better error handling and option for machineID

Author: aleksander-vedvik

broadcast.go

@@ -18,6 +18,7 @@ type broadcastServer struct {
 	viewMutex         sync.RWMutex
 	id                uint32
 	addr              string
+	machineID         uint64
 	view              RawConfiguration
 	createBroadcaster func(m BroadcastMetadata, o *BroadcastOrchestrator, b EnqueueBroadcast) Broadcaster
 	orchestrator      *BroadcastOrchestrator
@@ -29,9 +30,10 @@ func (srv *Server) GetStats() broadcast.Metrics {
 	return srv.broadcastSrv.manager.GetStats()
 }
 
-func newBroadcastServer(logger *slog.Logger, order map[string]int) *broadcastServer {
+func newBroadcastServer(logger *slog.Logger, order map[string]int, machineID uint64) *broadcastServer {
 	srv := &broadcastServer{
-		logger: logger,
+		logger:    logger,
+		machineID: machineID,
 	}
 	srv.manager = broadcast.NewBroadcastManager(logger, createClient, srv.canceler, order)
 	return srv
@@ -59,7 +61,7 @@ func (srv *broadcastServer) addAddr(addr string) {
 	h := fnv.New32a()
 	_, _ = h.Write([]byte(srv.addr))
 	srv.id = h.Sum32()
-	srv.manager.AddAddr(srv.id, srv.addr)
+	srv.manager.AddAddr(srv.id, srv.addr, srv.machineID)
 }
 
 const (

broadcast/manager.go

@@ -16,7 +16,7 @@ type Manager interface {
 	Cancel(uint64, []string) error
 	Done(uint64)
 	NewBroadcastID() uint64
-	AddAddr(id uint32, addr string)
+	AddAddr(id uint32, addr string, machineID uint64)
 	AddHandler(method string, handler any)
 	Close() error
 	ResetState()
@@ -149,10 +149,10 @@ func (mgr *manager) NewBroadcastID() uint64 {
 	return mgr.state.snowflake.NewBroadcastID()
 }
 
-func (mgr *manager) AddAddr(id uint32, addr string) {
+func (mgr *manager) AddAddr(id uint32, addr string, machineID uint64) {
 	mgr.router.id = id
 	mgr.router.addr = addr
-	mgr.state.snowflake = NewSnowflake(addr)
+	mgr.state.snowflake = NewSnowflake(machineID)
 }
 
 func (mgr *manager) AddHandler(method string, handler any) {

broadcast/processor.go

@@ -31,7 +31,7 @@ type metadata struct {
 	Sent              bool
 	ResponseMsg       protoreflect.ProtoMessage
 	ResponseErr       error
-	SendFn            func(protoreflect.ProtoMessage, error)
+	SendFn            func(protoreflect.ProtoMessage, error) error
 	IsBroadcastClient bool
 }
 
@@ -188,7 +188,10 @@ func (m *metadata) send(resp protoreflect.ProtoMessage, err error) error {
 	if !m.hasReceivedClientRequest() {
 		return MissingClientReqErr{}
 	}
-	m.SendFn(resp, err)
+	// error is intentionally ignored. We have not setup retry logic for failed
+	// deliveries to clients. Responding with nil will stop the broadcast request
+	// which is needed to prevent many stale goroutines.
+	_ = m.SendFn(resp, err)
 	return nil
 }
 

broadcast/processor_test.go

@@ -37,7 +37,7 @@ func (r *mockRouter) Send(broadcastID uint64, addr, method string, req any) erro
 func (r *mockRouter) Connect(addr string) {}*/
 
 func TestHandleBroadcastOption1(t *testing.T) {
-	snowflake := NewSnowflake("127.0.0.1:8080")
+	snowflake := NewSnowflake(0)
 	broadcastID := snowflake.NewBroadcastID()
 
 	var tests = []struct {
@@ -127,7 +127,7 @@ func TestHandleBroadcastOption1(t *testing.T) {
 	clientMsg := Content{
 		BroadcastID:       broadcastID,
 		IsBroadcastClient: true,
-		SendFn:            func(resp protoreflect.ProtoMessage, err error) {},
+		SendFn:            func(resp protoreflect.ProtoMessage, err error) error { return nil },
 		ReceiveChan:       make(chan shardResponse),
 	}
 	req.sendChan <- clientMsg
@@ -145,7 +145,7 @@ func TestHandleBroadcastOption1(t *testing.T) {
 }
 
 func TestHandleBroadcastCall1(t *testing.T) {
-	snowflake := NewSnowflake("127.0.0.1:8080")
+	snowflake := NewSnowflake(0)
 	broadcastID := snowflake.NewBroadcastID()
 
 	var tests = []struct {
@@ -249,7 +249,7 @@ func TestHandleBroadcastCall1(t *testing.T) {
 }
 
 func BenchmarkHandle1(b *testing.B) {
-	snowflake := NewSnowflake("127.0.0.1:8080")
+	snowflake := NewSnowflake(0)
 	originMethod := "testMethod"
 	router := &mockRouter{
 		returnError: false,
@@ -264,7 +264,7 @@ func BenchmarkHandle1(b *testing.B) {
 		},
 		BroadcastID: broadcastID,
 	}
-	sendFn := func(resp protoreflect.ProtoMessage, err error) {}
+	sendFn := func(resp protoreflect.ProtoMessage, err error) error { return nil }
 
 	b.ResetTimer()
 	b.Run("RequestHandler", func(b *testing.B) {

broadcast/request_test.go

@@ -38,7 +38,7 @@ func (r *mockRouter) Send(broadcastID uint64, addr, method string, req any) erro
 func (r *mockRouter) Connect(addr string) {}
 
 func TestHandleBroadcastOption(t *testing.T) {
-	snowflake := NewSnowflake("127.0.0.1:8080")
+	snowflake := NewSnowflake(0)
 	broadcastID := snowflake.NewBroadcastID()
 
 	var tests = []struct {
@@ -127,7 +127,7 @@ func TestHandleBroadcastOption(t *testing.T) {
 	clientMsg := Content{
 		BroadcastID:       broadcastID,
 		IsBroadcastClient: true,
-		SendFn:            func(resp protoreflect.ProtoMessage, err error) {},
+		SendFn:            func(resp protoreflect.ProtoMessage, err error) error { return nil },
 		ReceiveChan:       make(chan shardResponse),
 	}
 	req.sendChan <- clientMsg
@@ -145,7 +145,7 @@ func TestHandleBroadcastOption(t *testing.T) {
 }
 
 func TestHandleBroadcastCall(t *testing.T) {
-	snowflake := NewSnowflake("127.0.0.1:8080")
+	snowflake := NewSnowflake(0)
 	broadcastID := snowflake.NewBroadcastID()
 
 	var tests = []struct {
@@ -248,7 +248,7 @@ func TestHandleBroadcastCall(t *testing.T) {
 }
 
 func BenchmarkHandle2(b *testing.B) {
-	snowflake := NewSnowflake("127.0.0.1:8080")
+	snowflake := NewSnowflake(0)
 	originMethod := "testMethod"
 	router := &mockRouter{
 		returnError: false,
@@ -263,7 +263,7 @@ func BenchmarkHandle2(b *testing.B) {
 		},
 		BroadcastID: broadcastID,
 	}
-	sendFn := func(resp protoreflect.ProtoMessage, err error) {}
+	sendFn := func(resp protoreflect.ProtoMessage, err error) error { return nil }
 
 	b.ResetTimer()
 	b.Run("RequestHandler", func(b *testing.B) {

broadcast/shard.go

@@ -2,7 +2,6 @@ package broadcast
 
 import (
 	"context"
-	"log/slog"
 	"sync"
 	"time"
 )
@@ -126,7 +125,6 @@ func (s *shard) run(sendBuffer int) {
 				}
 			} else {
 				if msg.IsCancellation {
-					slog.Error("this is the reason")
 					// ignore cancellations if a broadcast request
 					// has not been created yet
 					continue

broadcast/shard_test.go

@@ -33,7 +33,7 @@ func (r *slowRouter) Send(broadcastID uint64, addr, method string, req any) erro
 func (r *slowRouter) Connect(addr string) {}
 
 func TestShard(t *testing.T) {
-	snowflake := NewSnowflake("127.0.0.1:8080")
+	snowflake := NewSnowflake(0)
 	broadcastID := snowflake.NewBroadcastID()
 	router := &slowRouter{
 		returnError: false,
@@ -151,7 +151,7 @@ func TestShard(t *testing.T) {
 }
 
 func BenchmarkShard(b *testing.B) {
-	snowflake := NewSnowflake("127.0.0.1:8080")
+	snowflake := NewSnowflake(0)
 	router := &slowRouter{
 		returnError: false,
 	}

broadcast/snowflake.go

@@ -1,6 +1,7 @@
 package broadcast
 
 import (
+	"log/slog"
 	"math/rand"
 	"sync"
 	"time"
@@ -16,8 +17,8 @@ type Snowflake struct {
 }
 
 const (
-	maxShard           = float32(1 << 4)
-	maxMachineID       = float32(1 << 12)
+	MaxMachineID       = uint16(1 << 12)
+	maxShard           = uint8(1 << 4)
 	maxSequenceNum     = uint32(1 << 18)
 	bitMaskTimestamp   = uint64((1<<30)-1) << 34
 	bitMaskShardID     = uint64((1<<4)-1) << 30
@@ -26,10 +27,15 @@ const (
 	epoch              = "2024-01-01T00:00:00"
 )
 
-func NewSnowflake(addr string) *Snowflake {
+func NewSnowflake(id uint64) *Snowflake {
+	if id < 0 || id >= uint64(MaxMachineID) {
+		newID := uint64(rand.Int31n(int32(MaxMachineID)))
+		slog.Warn("snowflakeID: provided machienID is higher than max or lower than min. A random ID will be assigned instead.", "max", MaxMachineID, "min", 0, "givenID", id, "assignedID", newID)
+		id = newID
+	}
 	timestamp, _ := time.Parse("2006-01-02T15:04:05", epoch)
 	return &Snowflake{
-		MachineID:   uint64(rand.Int31n(int32(maxMachineID))),
+		MachineID:   id,
 		SequenceNum: 0,
 		epoch:       timestamp,
 		//sequenceNum: uint32(maxSequenceNum * rand.Float32()),

broadcast/snowflake_test.go

@@ -0,0 +1,48 @@
+package broadcast
+
+import (
+	"testing"
+)
+
+func TestBroadcastID(t *testing.T) {
+	if MaxMachineID != 4096 {
+		t.Errorf("maxMachineID is hardcoded in test. want: %v, got: %v", 4096, MaxMachineID)
+	}
+	if maxSequenceNum != 262144 {
+		t.Errorf("maxSequenceNum is hardcoded in test. want: %v, got: %v", 262144, maxSequenceNum)
+	}
+	if maxShard != 16 {
+		t.Errorf("maxShard is hardcoded in test. want: %v, got: %v", 16, maxShard)
+	}
+	// intentionally provide an illegal machineID. A random machineID should be given instead.
+	snowflake := NewSnowflake(8000)
+	machineID := snowflake.MachineID
+	timestampDistribution := make(map[uint32]int)
+	maxN := 262144 // = 2^18
+	for j := 1; j < 3*maxN; j++ {
+		i := j % maxN
+		broadcastID := snowflake.NewBroadcastID()
+		timestamp, shard, m, n := DecodeBroadcastID(broadcastID)
+		if i != int(n) {
+			t.Errorf("wrong sequence number. want: %v, got: %v", i, n)
+		}
+		if m >= 4096 {
+			t.Errorf("machine ID cannot be higher than max. want: %v, got: %v", 4095, m)
+		}
+		if m != uint16(machineID) {
+			t.Errorf("wrong machine ID. want: %v, got: %v", machineID, m)
+		}
+		if shard >= 16 {
+			t.Errorf("cannot have higher shard than max. want: %v, got: %v", 15, shard)
+		}
+		if n >= uint32(maxN) {
+			t.Errorf("sequence number cannot be higher than max. want: %v, got: %v", maxN, n)
+		}
+		timestampDistribution[timestamp]++
+	}
+	for k, v := range timestampDistribution {
+		if v > maxN {
+			t.Errorf("cannot have more than maxN in a second. want: %v, got: %v", maxN, k)
+		}
+	}
+}

broadcast/state.go

@@ -160,7 +160,7 @@ type Content struct {
 	OriginMethod      string
 	CurrentMethod     string
 	ReceiveChan       chan shardResponse
-	SendFn            func(resp protoreflect.ProtoMessage, err error)
+	SendFn            func(resp protoreflect.ProtoMessage, err error) error
 	Ctx               context.Context
 	CancelCtx         context.CancelFunc
 	Run               func(context.Context, func(Msg) error)

handler.go

@@ -97,14 +97,14 @@ func createRequest(msg *broadcast.Content, ctx ServerCtx, in *Message, finished
 	}
 }
 
-func createSendFn(msgID uint64, method string, finished chan<- *Message, ctx ServerCtx) func(resp protoreflect.ProtoMessage, err error) {
-	return func(resp protoreflect.ProtoMessage, err error) {
+func createSendFn(msgID uint64, method string, finished chan<- *Message, ctx ServerCtx) func(resp protoreflect.ProtoMessage, err error) error {
+	return func(resp protoreflect.ProtoMessage, err error) error {
 		md := &ordering.Metadata{
 			MessageID: msgID,
 			Method:    method,
 		}
 		msg := WrapMessage(md, resp, err)
-		SendMessage(ctx, finished, msg)
+		return SendMessage(ctx, finished, msg)
 	}
 }
 

mgr.go

@@ -53,7 +53,7 @@ func NewRawManager(opts ...ManagerOption) *RawManager {
 	if m.logger != nil {
 		m.logger.Printf("ready")
 	}
-	m.snowflake = broadcast.NewSnowflake(m.opts.addr)
+	m.snowflake = broadcast.NewSnowflake(m.opts.machineID)
 	return m
 }
 

opts.go

@@ -19,7 +19,7 @@ type managerOptions struct {
 	metadata        metadata.MD
 	perNodeMD       func(uint32) metadata.MD
 	publicKey       string
-	addr            string
+	machineID       uint64
 }
 
 func newManagerOptions() managerOptions {
@@ -104,8 +104,8 @@ func WithPublicKey(publicKey string) ManagerOption {
 	}
 }
 
-func WithAddr(addr string) ManagerOption {
+func WithMachineID(id uint64) ManagerOption {
 	return func(o *managerOptions) {
-		o.addr = addr
+		o.machineID = id
 	}
 }

server.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"sync"
 
+	"github.com/relab/gorums/broadcast"
 	"github.com/relab/gorums/ordering"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -111,6 +112,13 @@ type serverOptions struct {
 	connectCallback func(context.Context)
 	logger          *slog.Logger
 	executionOrder  map[string]int
+	machineID       uint64
+}
+
+var defaultServerOptions = serverOptions{
+	// Provide an illegal machineID to avoid unintentional collisions.
+	// 0 is a valid MachineID and should not be used as default.
+	machineID: uint64(broadcast.MaxMachineID) + 1,
 }
 
 // ServerOption is used to change settings for the GorumsServer
@@ -167,14 +175,14 @@ type Server struct {
 // This function is intended for internal Gorums use.
 // You should call `NewServer` in the generated code instead.
 func NewServer(opts ...ServerOption) *Server {
-	var serverOpts serverOptions
+	serverOpts := defaultServerOptions
 	for _, opt := range opts {
 		opt(&serverOpts)
 	}
 	s := &Server{
 		srv:          newOrderingServer(&serverOpts),
 		grpcServer:   grpc.NewServer(serverOpts.grpcOpts...),
-		broadcastSrv: newBroadcastServer(serverOpts.logger, serverOpts.executionOrder),
+		broadcastSrv: newBroadcastServer(serverOpts.logger, serverOpts.executionOrder, serverOpts.machineID),
 	}
 	ordering.RegisterGorumsServer(s.grpcServer, s.srv)
 	return s