feat(broadcast): added more fields to broadcast metadata

Author: aleksander-vedvik

broadcast.go

@@ -170,12 +170,16 @@ type Broadcaster interface{}
 
 type BroadcastMetadata struct {
 	BroadcastID       uint64
-	IsBroadcastClient bool   // type of sender, could be: Client or Server
-	SenderAddr        string // address of last hop
-	OriginAddr        string // address of the origin
-	OriginMethod      string // the first method called by the origin
-	Method            string // the current method
-	Digest            []byte // digest of original message sent by client
+	IsBroadcastClient bool      // type of sender, could be: Client or Server
+	SenderAddr        string    // address of last hop
+	OriginAddr        string    // address of the origin
+	OriginMethod      string    // the first method called by the origin
+	Method            string    // the current method
+	Digest            []byte    // digest of original message sent by client
+	Timestamp         time.Time // timestamp in seconds when the broadcast request was issued by the client/server
+	ShardID           uint16    // ID of the shard handling the broadcast request
+	MachineID         uint16    // ID of the client/server that issued the broadcast request
+	SequenceNo        uint32    // sequence number of the broadcast request from that particular client/server. Will roll over when reaching max.
 }
 
 func newBroadcastMetadata(md *ordering.Metadata) BroadcastMetadata {
@@ -187,12 +191,17 @@ func newBroadcastMetadata(md *ordering.Metadata) BroadcastMetadata {
 	if len(tmp) >= 1 {
 		m = tmp[len(tmp)-1]
 	}
+	timestamp, shardID, machineID, sequenceNo := broadcast.DecodeBroadcastID(md.BroadcastMsg.BroadcastID)
 	return BroadcastMetadata{
 		BroadcastID:       md.BroadcastMsg.BroadcastID,
 		IsBroadcastClient: md.BroadcastMsg.IsBroadcastClient,
 		SenderAddr:        md.BroadcastMsg.SenderAddr,
 		OriginAddr:        md.BroadcastMsg.OriginAddr,
 		OriginMethod:      md.BroadcastMsg.OriginMethod,
 		Method:            m,
+		Timestamp:         broadcast.Epoch().Add(time.Duration(timestamp) * time.Second),
+		ShardID:           shardID,
+		MachineID:         machineID,
+		SequenceNo:        sequenceNo,
 	}
 }

broadcast/consts.go

@@ -19,6 +19,8 @@ type BroadcastOptions struct {
 
 const (
 	BroadcastID string = "broadcastID"
+	// special origin addr used in creating a broadcast request from a server
+	ServerOriginAddr string = "server"
 )
 
 type ServerHandler func(ctx context.Context, in protoreflect.ProtoMessage, broadcastID uint64, originAddr, originMethod string, options BroadcastOptions, id uint32, addr string)

broadcast/processor.go

@@ -8,18 +8,21 @@ import (
 )
 
 type BroadcastProcessor struct {
-	broadcastID           uint64
-	router                Router
-	broadcastChan         chan Msg
-	sendChan              chan Content
-	ctx                   context.Context
-	cancelFunc            context.CancelFunc
-	started               time.Time
-	ended                 time.Time
+	broadcastID   uint64
+	router        Router
+	broadcastChan chan Msg
+	sendChan      chan Content
+	ctx           context.Context
+	cancelFunc    context.CancelFunc
+	started       time.Time
+	ended         time.Time
+
+	// handled by shard
 	cancellationCtx       context.Context
-	cancellationCtxCancel context.CancelFunc // should only be called by the shard
+	cancellationCtxCancel context.CancelFunc
 	sentCancellation      bool
 
+	// ordering
 	executionOrder map[string]int
 	orderIndex     int
 	outOfOrderMsgs map[string][]Content

broadcast/router.go

@@ -102,7 +102,7 @@ func (r *BroadcastRouter) routeClientReply(broadcastID uint64, addr, method stri
 }
 
 func (r *BroadcastRouter) getClient(addr string) (*Client, error) {
-	if addr == "" {
+	if addr == "" || addr == ServerOriginAddr {
 		return nil, InvalidAddrErr{addr: addr}
 	}
 	// fast path:

broadcast/snowflake.go

@@ -26,18 +26,22 @@ const (
 	epoch              = "2024-01-01T00:00:00"
 )
 
+func Epoch() time.Time {
+	timestamp, _ := time.Parse("2006-01-02T15:04:05", epoch)
+	return timestamp
+}
+
 func NewSnowflake(id uint64) *Snowflake {
 	if id < 0 || id >= uint64(MaxMachineID) {
 		//newID := uint64(rand.Int31n(int32(MaxMachineID)))
 		//slog.Warn("snowflakeID: provided machineID 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
 		id = uint64(rand.Int31n(int32(MaxMachineID)))
 	}
-	timestamp, _ := time.Parse("2006-01-02T15:04:05", epoch)
 	return &Snowflake{
 		MachineID:   id,
 		SequenceNum: 0,
-		epoch:       timestamp,
+		epoch:       Epoch(),
 		//sequenceNum: uint32(maxSequenceNum * rand.Float32()),
 	}
 }
@@ -70,7 +74,7 @@ start:
 	return t | shard | m | n
 }
 
-func DecodeBroadcastID(broadcastID uint64) (uint32, uint16, uint16, uint32) {
+func DecodeBroadcastID(broadcastID uint64) (timestamp uint32, shardID uint16, machineID uint16, sequenceNo uint32) {
 	t := (broadcastID & bitMaskTimestamp) >> 34
 	shard := (broadcastID & bitMaskShardID) >> 30
 	m := (broadcastID & bitMaskMachineID) >> 18

handler.go

@@ -180,7 +180,7 @@ func (srv *broadcastServer) serverBroadcastHandler(method string, req protorefle
 		Message:           req,
 		Method:            method,
 		BroadcastID:       srv.manager.NewBroadcastID(),
-		OriginAddr:        "server",
+		OriginAddr:        broadcast.ServerOriginAddr,
 		IsBroadcastClient: false,
 	}
 	srv.viewMutex.RLock()