Only broadcast actions from api (#4643)

---------

Co-authored-by: CoderZhi <[email protected]>

Author: envestcc

actpool/actpool.go

@@ -85,7 +85,7 @@ type ActPool interface {
 
 // Subscriber is the interface for actpool subscriber
 type Subscriber interface {
-	OnAdded(*action.SealedEnvelope)
+	OnAdded(context.Context, *action.SealedEnvelope)
 	OnRemoved(*action.SealedEnvelope)
 }
 
@@ -547,9 +547,9 @@ func (ap *actPool) AddSubscriber(sub Subscriber) {
 	ap.subs = append(ap.subs, sub)
 }
 
-func (ap *actPool) onAdded(act *action.SealedEnvelope) {
+func (ap *actPool) onAdded(ctx context.Context, act *action.SealedEnvelope) {
 	for _, sub := range ap.subs {
-		sub.OnAdded(act)
+		sub.OnAdded(ctx, act)
 	}
 }
 

actpool/queueworker.go

@@ -105,7 +105,7 @@ func (worker *queueWorker) Handle(job workerJob) error {
 	}
 
 	worker.ap.allActions.Set(actHash, act)
-	worker.ap.onAdded(act)
+	worker.ap.onAdded(ctx, act)
 	isBlobTx := len(act.BlobHashes()) > 0 // only store blob tx
 	if worker.ap.store != nil && isBlobTx {
 		if err := worker.ap.store.Put(act); err != nil {

actpool/validator.go

@@ -38,7 +38,7 @@ func (v *blobValidator) Validate(ctx context.Context, act *action.SealedEnvelope
 	return nil
 }
 
-func (v *blobValidator) OnAdded(act *action.SealedEnvelope) {
+func (v *blobValidator) OnAdded(ctx context.Context, act *action.SealedEnvelope) {
 	if len(act.BlobHashes()) == 0 {
 		return
 	}

api/action_radio.go

@@ -61,7 +61,11 @@ func (ar *ActionRadio) Stop() error {
 }
 
 // OnAdded broadcasts the action to the network
-func (ar *ActionRadio) OnAdded(selp *action.SealedEnvelope) {
+func (ar *ActionRadio) OnAdded(ctx context.Context, selp *action.SealedEnvelope) {
+	if _, fromAPI := GetAPIContext(ctx); !fromAPI {
+		// only broadcast actions from API context
+		return
+	}
 	var (
 		hasSidecar = selp.BlobTxSidecar() != nil
 		hash, _    = selp.Hash()

api/action_radio_test.go

@@ -32,6 +32,9 @@ func TestActionRadio(t *testing.T) {
 	selp, err := action.SignedTransfer(identityset.Address(1).String(), identityset.PrivateKey(1), 1, big.NewInt(1), nil, gas, gasPrice)
 	r.NoError(err)
 
-	radio.OnAdded(selp)
+	radio.OnAdded(WithAPIContext(context.Background()), selp)
+	r.Equal(uint64(1), atomic.LoadUint64(&broadcastCount))
+
+	radio.OnAdded(context.Background(), selp)
 	r.Equal(uint64(1), atomic.LoadUint64(&broadcastCount))
 }

api/context.go

@@ -12,6 +12,8 @@ type (
 		listenerIDs map[string]struct{}
 		mutex       sync.Mutex
 	}
+
+	apiContextKey struct{}
 )
 
 func (sc *StreamContext) AddListener(id string) {
@@ -46,3 +48,15 @@ func StreamFromContext(ctx context.Context) (*StreamContext, bool) {
 	sc, ok := ctx.Value(streamContextKey{}).(*StreamContext)
 	return sc, ok
 }
+
+func WithAPIContext(ctx context.Context) context.Context {
+	return context.WithValue(ctx, apiContextKey{}, struct{}{})
+}
+
+func GetAPIContext(ctx context.Context) (struct{}, bool) {
+	c := ctx.Value(apiContextKey{})
+	if c == nil {
+		return struct{}{}, false
+	}
+	return c.(struct{}), true
+}

api/coreservice.go

@@ -489,6 +489,7 @@ func (core *coreService) SendAction(ctx context.Context, in *iotextypes.Action)
 		return "", err
 	}
 	l := log.T(ctx).Logger().With(zap.String("actionHash", hex.EncodeToString(hash[:])))
+	ctx = WithAPIContext(ctx)
 	if err = core.ap.Add(ctx, selp); err != nil {
 		txBytes, serErr := proto.Marshal(in)
 		if serErr != nil {