refact: memberslist CallbackBroadcast

Author: soonkuk

go.mod

@@ -91,4 +91,4 @@ require (
 	golang.org/x/tools v0.36.0 // indirect
 )
 
-replace github.com/hashicorp/memberlist => github.com/spikeekips/memberlist v0.0.0-20230626195851-39f17fa10d23 // latest fix-data-race branch
+replace github.com/hashicorp/memberlist => github.com/HayoungOh5/memberlist v0.0.0-20251120091718-913bc68ce0d2

go.sum

@@ -1,4 +1,6 @@
 github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
+github.com/HayoungOh5/memberlist v0.0.0-20251120091718-913bc68ce0d2 h1:bbPmccnva4F5KQOkUwOsBqYKTDvSCXQu8XV+cKqGEG8=
+github.com/HayoungOh5/memberlist v0.0.0-20251120091718-913bc68ce0d2/go.mod h1:yvyXLpo0QaGE59Y7hDTsTzDD25JYBZ4mHgHUZ8lrOI0=
 github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0=
 github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM=
 github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII=
@@ -283,8 +285,6 @@ github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 h1:nn5Wsu0esKSJiIVhscUt
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
-github.com/spikeekips/memberlist v0.0.0-20230626195851-39f17fa10d23 h1:nE7dp89DUO4ETbLLwU93ICJPox08LI+yB5UbRx+nHtE=
-github.com/spikeekips/memberlist v0.0.0-20230626195851-39f17fa10d23/go.mod h1:yvyXLpo0QaGE59Y7hDTsTzDD25JYBZ4mHgHUZ8lrOI0=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=

launch/p_states_network_handlers.go

@@ -165,7 +165,7 @@ func AttachHandlerSendOperation(pctx context.Context) error {
 					}
 				}
 
-				return memberlist.CallbackBroadcast(b, id, nil)
+				return memberlist.CallbackBroadcast2(b, id, nil)
 			},
 			params.MISC.MaxMessageSize,
 		),

network/quicmemberlist/delegate.go

@@ -68,6 +68,10 @@ func (d *Delegate) QueueBroadcast(b memberlist.Broadcast) {
 	d.qu.QueueBroadcast(b)
 }
 
+func (d *Delegate) QueueBroadcast2(b memberlist.Broadcast) {
+	d.qu.QueueBroadcast2(b)
+}
+
 func (d *Delegate) GetBroadcasts(overhead, limit int) [][]byte {
 	d.qu.RetransmitMult = d.reTransmitMult()
 	return d.qu.GetBroadcasts(overhead, limit)

network/quicmemberlist/memberlist.go

@@ -323,6 +323,22 @@ func (srv *Memberlist) Broadcast(b memberlist.Broadcast) {
 	srv.delegate.QueueBroadcast(b)
 }
 
+func (srv *Memberlist) Broadcast2(b memberlist.Broadcast) {
+	if !srv.CanBroadcast() {
+		b.Finished()
+
+		return
+	}
+
+	if srv.metricsCollector != nil {
+		srv.metricsCollector.RecordMemberlistBroadcast()
+	}
+
+	srv.Log().Trace().Interface("broadcast", b).Msg("enqueue broadcast")
+
+	srv.delegate.QueueBroadcast2(b)
+}
+
 func (srv *Memberlist) CallbackBroadcast(b []byte, id string, notifych chan struct{}) error {
 	if !srv.CanBroadcast() {
 		if notifych != nil {
@@ -352,6 +368,35 @@ func (srv *Memberlist) CallbackBroadcast(b []byte, id string, notifych chan stru
 	return nil
 }
 
+func (srv *Memberlist) CallbackBroadcast2(b []byte, id string, notifych chan struct{}) error {
+	if !srv.CanBroadcast() {
+		if notifych != nil {
+			close(notifych)
+		}
+
+		return nil
+	}
+
+	// NOTE save b in cache first
+	srv.cbcache.Set(id, b, srv.args.CallbackBroadcastMessageExpire)
+
+	switch i, err := srv.args.Encoder.Marshal(
+		NewConnInfoBroadcastMessage(id, srv.local.ConnInfo())); {
+	case err != nil:
+		return err
+	default:
+		buf := bytes.NewBuffer(nil)
+		defer buf.Reset()
+
+		_, _ = buf.Write(callbackBroadcastMessageHeaderPrefix)
+		_, _ = buf.Write(i)
+
+		srv.Broadcast2(NewBroadcast(buf.Bytes(), id, notifych))
+	}
+
+	return nil
+}
+
 func (srv *Memberlist) CallbackBroadcastHandler() quicstreamheader.Handler[CallbackBroadcastMessageHeader] {
 	var sg singleflight.Group