fix allowing client negotiations prematurely.

cmd/cli/client/client.go

@@ -127,7 +127,8 @@ func NewRTCClient(conn *websocket.Conn) (*RTCClient, error) {
 	})
 
 	peerConn.OnTrack(func(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) {
-		logger.Debugw("track received", "label", track.StreamID(), "id", track.ID())
+		logger.Debugw("track received", "label", track.StreamID(), "id", track.ID(),
+			"participant", c.localParticipant.Identity)
 		go c.processTrack(track)
 	})
 
@@ -499,7 +500,7 @@ func (c *RTCClient) handleAnswer(desc webrtc.SessionDescription) error {
 }
 
 func (c *RTCClient) requestNegotiation() error {
-	logger.Debugw("requesting negotiation")
+	logger.Debugw("requesting negotiation", "participant", c.localParticipant.Identity)
 	return c.SendRequest(&livekit.SignalRequest{
 		Message: &livekit.SignalRequest_Negotiate{
 			Negotiate: &livekit.NegotiationRequest{},
@@ -508,7 +509,7 @@ func (c *RTCClient) requestNegotiation() error {
 }
 
 func (c *RTCClient) negotiate() error {
-	logger.Debugw("starting negotiation")
+	logger.Debugw("starting negotiation", "participant", c.localParticipant.Identity)
 	offer, err := c.PeerConn.CreateOffer(nil)
 	if err != nil {
 		return err

pkg/routing/interfaces.go

@@ -33,6 +33,7 @@ type Router interface {
 	ClearRoomState(roomName string) error
 	RegisterNode() error
 	UnregisterNode() error
+	RemoveDeadNodes() error
 	GetNode(nodeId string) (*livekit.Node, error)
 	ListNodes() ([]*livekit.Node, error)
 

pkg/routing/localrouter.go

@@ -46,6 +46,10 @@ func (r *LocalRouter) UnregisterNode() error {
 	return nil
 }
 
+func (r *LocalRouter) RemoveDeadNodes() error {
+	return nil
+}
+
 func (r *LocalRouter) GetNode(nodeId string) (*livekit.Node, error) {
 	if nodeId == r.currentNode.Id {
 		return r.currentNode, nil

pkg/routing/redis.go

@@ -58,6 +58,9 @@ func publishRTCMessage(rc *redis.Client, nodeId string, participantKey string, m
 	if err != nil {
 		return err
 	}
+
+	//logger.Debugw("publishing to", "rtcChannel", rtcNodeChannel(nodeId),
+	//	"message", rm.Message)
 	return rc.Publish(redisCtx, rtcNodeChannel(nodeId), data).Err()
 }
 

pkg/routing/redisrouter.go

@@ -16,7 +16,7 @@ import (
 const (
 	// expire participant mappings after a day
 	participantMappingTTL = 24 * time.Hour
-	statsUpdateInterval   = 10 * time.Second
+	statsUpdateInterval   = 2 * time.Second
 )
 
 // RedisRouter uses Redis pub/sub to route signaling messages across different nodes
@@ -25,7 +25,6 @@ const (
 type RedisRouter struct {
 	LocalRouter
 	rc        *redis.Client
-	cr        *utils.CachedRedis
 	ctx       context.Context
 	isStarted utils.AtomicFlag
 
@@ -46,7 +45,6 @@ func NewRedisRouter(currentNode LocalNode, rc *redis.Client) *RedisRouter {
 		signalSinks: make(map[string]*SignalNodeSink),
 	}
 	rr.ctx, rr.cancel = context.WithCancel(context.Background())
-	rr.cr = utils.NewCachedRedis(rr.ctx, rr.rc)
 	return rr
 }
 
@@ -55,7 +53,6 @@ func (r *RedisRouter) RegisterNode() error {
 	if err != nil {
 		return err
 	}
-	r.cr.ExpireHash(NodesKey, r.currentNode.Id)
 	if err := r.rc.HSet(r.ctx, NodesKey, r.currentNode.Id, data).Err(); err != nil {
 		return errors.Wrap(err, "could not register node")
 	}
@@ -67,8 +64,23 @@ func (r *RedisRouter) UnregisterNode() error {
 	return r.rc.HDel(context.Background(), NodesKey, r.currentNode.Id).Err()
 }
 
+func (r *RedisRouter) RemoveDeadNodes() error {
+	nodes, err := r.ListNodes()
+	if err != nil {
+		return err
+	}
+	for _, n := range nodes {
+		if !IsAvailable(n) {
+			if err := r.rc.HDel(context.Background(), NodesKey, n.Id).Err(); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 func (r *RedisRouter) GetNodeForRoom(roomName string) (string, error) {
-	val, err := r.cr.CachedHGet(NodeRoomKey, roomName)
+	val, err := r.rc.HGet(r.ctx, NodeRoomKey, roomName).Result()
 	if err != nil {
 		err = errors.Wrap(err, "could not get node for room")
 	}
@@ -87,7 +99,7 @@ func (r *RedisRouter) ClearRoomState(roomName string) error {
 }
 
 func (r *RedisRouter) GetNode(nodeId string) (*livekit.Node, error) {
-	data, err := r.cr.CachedHGet(NodesKey, nodeId)
+	data, err := r.rc.HGet(r.ctx, NodesKey, nodeId).Result()
 	if err != nil {
 		return nil, err
 	}
@@ -221,7 +233,6 @@ func (r *RedisRouter) Stop() {
 }
 
 func (r *RedisRouter) setParticipantRTCNode(participantKey, nodeId string) error {
-	r.cr.Expire(participantRTCKey(participantKey))
 	err := r.rc.Set(r.ctx, participantRTCKey(participantKey), nodeId, participantMappingTTL).Err()
 	if err != nil {
 		err = errors.Wrap(err, "could not set rtc node")
@@ -230,7 +241,6 @@ func (r *RedisRouter) setParticipantRTCNode(participantKey, nodeId string) error
 }
 
 func (r *RedisRouter) setParticipantSignalNode(connectionId, nodeId string) error {
-	r.cr.Expire(participantSignalKey(connectionId))
 	if err := r.rc.Set(r.ctx, participantSignalKey(connectionId), nodeId, participantMappingTTL).Err(); err != nil {
 		return errors.Wrap(err, "could not set signal node")
 	}
@@ -276,17 +286,17 @@ func (r *RedisRouter) getOrCreateSignalSink(nodeId string, connectionId string)
 }
 
 func (r *RedisRouter) getParticipantRTCNode(participantKey string) (string, error) {
-	return r.cr.CachedGet(participantRTCKey(participantKey))
+	return r.rc.Get(r.ctx, participantRTCKey(participantKey)).Result()
 }
 
 func (r *RedisRouter) getParticipantSignalNode(connectionId string) (nodeId string, err error) {
-	return r.cr.CachedGet(participantSignalKey(connectionId))
+	return r.rc.Get(r.ctx, participantSignalKey(connectionId)).Result()
 }
 
 // update node stats and cleanup
 func (r *RedisRouter) statsWorker() {
 	for r.ctx.Err() == nil {
-		// update every 10 seconds
+		// update periodically seconds
 		select {
 		case <-time.After(statsUpdateInterval):
 			r.currentNode.Stats.UpdatedAt = time.Now().Unix()
@@ -343,6 +353,9 @@ func (r *RedisRouter) handleSignalMessage(sm *livekit.SignalNodeMessage) error {
 
 	switch rmb := sm.Message.(type) {
 	case *livekit.SignalNodeMessage_Response:
+		//logger.Debugw("forwarding signal message",
+		//	"connectionId", connectionId,
+		//	"type", fmt.Sprintf("%T", rmb.Response.Message))
 		// in the event the current node is an Signal node, push to response channels
 		resSink := r.getOrCreateMessageChannel(r.responseChannels, connectionId)
 		if err := resSink.WriteMessage(rmb.Response); err != nil {
@@ -350,7 +363,8 @@ func (r *RedisRouter) handleSignalMessage(sm *livekit.SignalNodeMessage) error {
 		}
 
 	case *livekit.SignalNodeMessage_EndSession:
-		logger.Debugw("received EndSession, closing signal connection")
+		logger.Debugw("received EndSession, closing signal connection",
+			"connectionId", connectionId)
 		resSink := r.getOrCreateMessageChannel(r.responseChannels, connectionId)
 		resSink.Close()
 	}

pkg/routing/routingfakes/fake_router.go

@@ -73,6 +73,16 @@ type FakeRouter struct {
 	registerNodeReturnsOnCall map[int]struct {
 		result1 error
 	}
+	RemoveDeadNodesStub        func() error
+	removeDeadNodesMutex       sync.RWMutex
+	removeDeadNodesArgsForCall []struct {
+	}
+	removeDeadNodesReturns struct {
+		result1 error
+	}
+	removeDeadNodesReturnsOnCall map[int]struct {
+		result1 error
+	}
 	SetNodeForRoomStub        func(string, string) error
 	setNodeForRoomMutex       sync.RWMutex
 	setNodeForRoomArgsForCall []struct {
@@ -460,6 +470,59 @@ func (fake *FakeRouter) RegisterNodeReturnsOnCall(i int, result1 error) {
 	}{result1}
 }
 
+func (fake *FakeRouter) RemoveDeadNodes() error {
+	fake.removeDeadNodesMutex.Lock()
+	ret, specificReturn := fake.removeDeadNodesReturnsOnCall[len(fake.removeDeadNodesArgsForCall)]
+	fake.removeDeadNodesArgsForCall = append(fake.removeDeadNodesArgsForCall, struct {
+	}{})
+	stub := fake.RemoveDeadNodesStub
+	fakeReturns := fake.removeDeadNodesReturns
+	fake.recordInvocation("RemoveDeadNodes", []interface{}{})
+	fake.removeDeadNodesMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeRouter) RemoveDeadNodesCallCount() int {
+	fake.removeDeadNodesMutex.RLock()
+	defer fake.removeDeadNodesMutex.RUnlock()
+	return len(fake.removeDeadNodesArgsForCall)
+}
+
+func (fake *FakeRouter) RemoveDeadNodesCalls(stub func() error) {
+	fake.removeDeadNodesMutex.Lock()
+	defer fake.removeDeadNodesMutex.Unlock()
+	fake.RemoveDeadNodesStub = stub
+}
+
+func (fake *FakeRouter) RemoveDeadNodesReturns(result1 error) {
+	fake.removeDeadNodesMutex.Lock()
+	defer fake.removeDeadNodesMutex.Unlock()
+	fake.RemoveDeadNodesStub = nil
+	fake.removeDeadNodesReturns = struct {
+		result1 error
+	}{result1}
+}
+
+func (fake *FakeRouter) RemoveDeadNodesReturnsOnCall(i int, result1 error) {
+	fake.removeDeadNodesMutex.Lock()
+	defer fake.removeDeadNodesMutex.Unlock()
+	fake.RemoveDeadNodesStub = nil
+	if fake.removeDeadNodesReturnsOnCall == nil {
+		fake.removeDeadNodesReturnsOnCall = make(map[int]struct {
+			result1 error
+		})
+	}
+	fake.removeDeadNodesReturnsOnCall[i] = struct {
+		result1 error
+	}{result1}
+}
+
 func (fake *FakeRouter) SetNodeForRoom(arg1 string, arg2 string) error {
 	fake.setNodeForRoomMutex.Lock()
 	ret, specificReturn := fake.setNodeForRoomReturnsOnCall[len(fake.setNodeForRoomArgsForCall)]
@@ -736,6 +799,8 @@ func (fake *FakeRouter) Invocations() map[string][][]interface{} {
 	defer fake.onNewParticipantRTCMutex.RUnlock()
 	fake.registerNodeMutex.RLock()
 	defer fake.registerNodeMutex.RUnlock()
+	fake.removeDeadNodesMutex.RLock()
+	defer fake.removeDeadNodesMutex.RUnlock()
 	fake.setNodeForRoomMutex.RLock()
 	defer fake.setNodeForRoomMutex.RUnlock()
 	fake.startMutex.RLock()

pkg/routing/utils_test.go

@@ -14,7 +14,7 @@ func TestIsAvailable(t *testing.T) {
 	t.Run("still available", func(t *testing.T) {
 		n := &livekit.Node{
 			Stats: &livekit.NodeStats{
-				UpdatedAt: time.Now().Unix() - 10,
+				UpdatedAt: time.Now().Unix() - 3,
 			},
 		}
 		assert.True(t, routing.IsAvailable(n))

pkg/rtc/participant.go

@@ -113,7 +113,8 @@ func NewParticipant(identity string, pc types.PeerConnection, rs routing.Message
 			},
 		})
 		if err != nil {
-			logger.Errorw("could not send trickle", "err", err)
+			logger.Errorw("could not send trickle", "err", err,
+				"participant", identity)
 		}
 
 		if participant.onICECandidate != nil {
@@ -246,7 +247,7 @@ func (p *ParticipantImpl) Answer(sdp webrtc.SessionDescription) (answer webrtc.S
 	}
 	p.negotiationCond.L.Unlock()
 
-	logger.Debugw("sending to client answer",
+	logger.Debugw("sending answer to client",
 		"participant", p.Identity(),
 	//"sdp", sdp.SDP,
 	)
@@ -279,7 +280,7 @@ func (p *ParticipantImpl) AddTrack(clientId, name string, trackType livekit.Trac
 	}
 	p.pendingTracks[clientId] = ti
 
-	p.responseSink.WriteMessage(&livekit.SignalResponse{
+	err := p.responseSink.WriteMessage(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_TrackPublished{
 			TrackPublished: &livekit.TrackPublishedResponse{
 				Cid:   clientId,
@@ -287,6 +288,10 @@ func (p *ParticipantImpl) AddTrack(clientId, name string, trackType livekit.Trac
 			},
 		},
 	})
+	if err != nil {
+		logger.Errorw("could not write message", "error", err,
+			"participant", p.identity)
+	}
 }
 
 func (p *ParticipantImpl) HandleAnswer(sdp webrtc.SessionDescription) error {
@@ -323,11 +328,15 @@ func (p *ParticipantImpl) HandleClientNegotiation() {
 
 	logger.Debugw("allowing participant to negotiate",
 		"participant", p.Identity())
-	p.responseSink.WriteMessage(&livekit.SignalResponse{
+	err := p.responseSink.WriteMessage(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_Negotiate{
 			Negotiate: &livekit.NegotiationResponse{},
 		},
 	})
+	if err != nil {
+		logger.Errorw("could not write message", "error", err,
+			"participant", p.identity)
+	}
 }
 
 // AddICECandidate adds candidates for remote peer
@@ -369,11 +378,16 @@ func (p *ParticipantImpl) Close() error {
 func (p *ParticipantImpl) AddSubscriber(op types.Participant) error {
 	p.lock.RLock()
 	tracks := funk.Values(p.publishedTracks).([]types.PublishedTrack)
-	defer p.lock.RUnlock()
+	p.lock.RUnlock()
+
+	if len(tracks) == 0 {
+		return nil
+	}
 
 	logger.Debugw("subscribing new participant to tracks",
 		"srcParticipant", p.Identity(),
-		"dstParticipant", op.Identity())
+		"newParticipant", op.Identity(),
+		"numTracks", len(tracks))
 
 	for _, track := range tracks {
 		if err := track.AddSubscriber(op); err != nil {
@@ -472,13 +486,13 @@ func (p *ParticipantImpl) negotiate() {
 		return
 	}
 
-	logger.Debugw("starting server negotiation", "participant", p.Identity())
 	p.negotiationCond.L.Lock()
 	for p.negotiationState != negotiationStateNone {
 		p.negotiationCond.Wait()
-		p.negotiationState = negotiationStateServer
 	}
+	p.negotiationState = negotiationStateServer
 	p.negotiationCond.L.Unlock()
+	logger.Debugw("starting server negotiation", "participant", p.Identity())
 
 	offer, err := p.peerConn.CreateOffer(nil)
 	if err != nil {
@@ -508,8 +522,9 @@ func (p *ParticipantImpl) negotiate() {
 		},
 	})
 	if err != nil {
-		logger.Errorw("could not send offer to peer",
-			"err", err)
+		logger.Errorw("could not send offer to participant",
+			"err", err,
+			"participant", p.identity)
 	}
 }
 

pkg/service/roommanager.go

@@ -253,7 +253,7 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici
 			case *livekit.SignalRequest_Offer:
 				_, err := participant.Answer(rtc.FromProtoSessionDescription(msg.Offer))
 				if err != nil {
-					logger.Errorw("could not handle join", "err", err, "participant", participant.ID())
+					logger.Errorw("could not handle offer", "err", err, "participant", participant.Identity())
 					return
 				}
 			case *livekit.SignalRequest_AddTrack:
@@ -262,13 +262,13 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici
 				participant.AddTrack(msg.AddTrack.Cid, msg.AddTrack.Name, msg.AddTrack.Type)
 			case *livekit.SignalRequest_Answer:
 				if participant.State() == livekit.ParticipantInfo_JOINING {
-					logger.Errorw("cannot negotiate before peer offer", "participant", participant.ID())
+					logger.Errorw("cannot negotiate before peer offer", "participant", participant.Identity())
 					//conn.WriteJSON(jsonError(http.StatusNotAcceptable, "cannot negotiate before peer offer"))
 					return
 				}
 				sd := rtc.FromProtoSessionDescription(msg.Answer)
 				if err := participant.HandleAnswer(sd); err != nil {
-					logger.Errorw("could not handle answer", "participant", participant.ID(), "err", err)
+					logger.Errorw("could not handle answer", "participant", participant.Identity(), "err", err)
 					//conn.WriteJSON(
 					//	jsonError(http.StatusInternalServerError, "could not handle negotiate", err.Error()))
 					return
@@ -277,7 +277,7 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici
 				participant.HandleClientNegotiation()
 			case *livekit.SignalRequest_Trickle:
 				if participant.State() == livekit.ParticipantInfo_JOINING {
-					logger.Errorw("cannot trickle before peer offer", "participant", participant.ID())
+					logger.Errorw("cannot trickle before offer", "participant", participant.Identity())
 					//conn.WriteJSON(jsonError(http.StatusNotAcceptable, "cannot trickle before peer offer"))
 					return
 				}
@@ -285,7 +285,7 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici
 				candidateInit := rtc.FromProtoTrickle(msg.Trickle)
 				//logger.Debugw("adding peer candidate", "participant", participant.ID())
 				if err := participant.AddICECandidate(candidateInit); err != nil {
-					logger.Errorw("could not handle trickle", "participant", participant.ID(), "err", err)
+					logger.Errorw("could not handle trickle", "participant", participant.Identity(), "err", err)
 					//conn.WriteJSON(
 					//	jsonError(http.StatusInternalServerError, "could not handle trickle", err.Error()))
 					return

pkg/service/rtcservice.go

@@ -123,7 +123,9 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 				}
 				res, ok := msg.(*livekit.SignalResponse)
 				if !ok {
-					logger.Errorw("unexpected message type", "type", fmt.Sprintf("%T", msg))
+					logger.Errorw("unexpected message type",
+						"type", fmt.Sprintf("%T", msg),
+						"participant", identity)
 					continue
 				}
 

pkg/service/server.go

@@ -69,7 +69,12 @@ func NewLivekitServer(conf *config.Config,
 	router.OnNewParticipantRTC(roomManager.StartSession)
 
 	// clean up old rooms on startup
-	err = roomManager.CleanupRooms()
+	if err = roomManager.CleanupRooms(); err != nil {
+		return
+	}
+	if err = router.RemoveDeadNodes(); err != nil {
+		return
+	}
 
 	return
 }

test/integration_helpers.go

@@ -17,6 +17,7 @@ import (
 	"github.com/livekit/livekit-server/pkg/logger"
 	"github.com/livekit/livekit-server/pkg/routing"
 	"github.com/livekit/livekit-server/pkg/service"
+	"github.com/livekit/livekit-server/pkg/utils"
 	"github.com/livekit/livekit-server/proto/livekit"
 )
 
@@ -61,8 +62,8 @@ func setupSingleNodeTest(roomName string) *service.LivekitServer {
 }
 
 func setupMultiNodeTest() (*service.LivekitServer, *service.LivekitServer) {
-	s1 := createMultiNodeServer(nodeId1, defaultServerPort)
-	s2 := createMultiNodeServer(nodeId2, secondServerPort)
+	s1 := createMultiNodeServer(utils.NewGuid(nodeId1), defaultServerPort)
+	s2 := createMultiNodeServer(utils.NewGuid(nodeId2), secondServerPort)
 	go s1.Start()
 	go s2.Start()
 
@@ -153,7 +154,7 @@ func createSingleNodeServer() *service.LivekitServer {
 	}
 
 	currentNode, err := routing.NewLocalNode(conf)
-	currentNode.Id = nodeId1
+	currentNode.Id = utils.NewGuid(nodeId1)
 	if err != nil {
 		panic(err)
 	}
@@ -161,6 +162,7 @@ func createSingleNodeServer() *service.LivekitServer {
 	// local routing and store
 	router := routing.NewLocalRouter(currentNode)
 	roomStore := service.NewLocalRoomStore()
+	roomStore.DeleteRoom(testRoom)
 	s, err := service.InitializeServer(conf, &StaticKeyProvider{}, roomStore, router, currentNode, &routing.RandomSelector{})
 	if err != nil {
 		panic(fmt.Sprintf("could not create server: %v", err))
@@ -190,9 +192,13 @@ func createMultiNodeServer(nodeId string, port uint32) *service.LivekitServer {
 	if err = rc.Ping(context.Background()).Err(); err != nil {
 		panic(err)
 	}
+	if err = rc.Del(context.Background(), routing.NodesKey).Err(); err != nil {
+		panic(err)
+	}
 
 	router := routing.NewRedisRouter(currentNode, rc)
 	roomStore := service.NewRedisRoomStore(rc)
+	roomStore.DeleteRoom(testRoom)
 	s, err := service.InitializeServer(conf, &StaticKeyProvider{}, roomStore, router, currentNode, &routing.RandomSelector{})
 	if err != nil {
 		panic(fmt.Sprintf("could not create server: %v", err))

test/multinode_test.go

@@ -15,7 +15,7 @@ func TestMultiNodeRouting(t *testing.T) {
 		return
 	}
 
-	logger.Infow("---Starting TestMultiNodeRouting---")
+	logger.Infow("\n\n---Starting TestMultiNodeRouting---")
 	defer logger.Infow("---Finishing TestMultiNodeRouting---")
 
 	s1, s2 := setupMultiNodeTest()
@@ -24,8 +24,7 @@ func TestMultiNodeRouting(t *testing.T) {
 
 	// creating room on node 1
 	_, err := roomClient.CreateRoom(contextWithCreateRoomToken(), &livekit.CreateRoomRequest{
-		Name:   testRoom,
-		NodeId: nodeId1,
+		Name: testRoom,
 	})
 	assert.NoError(t, err)
 
@@ -80,7 +79,7 @@ func TestConnectWithoutCreation(t *testing.T) {
 		t.SkipNow()
 		return
 	}
-	logger.Infow("---Starting TestConnectWithoutCreation---")
+	logger.Infow("\n\n---Starting TestConnectWithoutCreation---")
 	defer logger.Infow("---Finishing TestConnectWithoutCreation---")
 
 	s1, s2 := setupMultiNodeTest()
@@ -100,7 +99,7 @@ func TestMultinodePublishingUponJoining(t *testing.T) {
 		return
 	}
 
-	logger.Infow("---Starting TestMultinodePublishingUponJoining---")
+	logger.Infow("\n\n---Starting TestMultinodePublishingUponJoining---")
 	defer logger.Infow("---Finishing TestMultinodePublishingUponJoining---")
 
 	s1, s2 := setupMultiNodeTest()
@@ -116,7 +115,7 @@ func TestMultinodeReceiveBeforePublish(t *testing.T) {
 		return
 	}
 
-	logger.Infow("---Starting TestMultinodeReceiveBeforePublish---")
+	logger.Infow("\n\n---Starting TestMultinodeReceiveBeforePublish---")
 	defer logger.Infow("---Finishing TestMultinodeReceiveBeforePublish---")
 
 	s1, s2 := setupMultiNodeTest()