Function TestNotifyMsgResendsOnlyChanges

kv/memberlist/memberlist_client_test.go:1720–1785

(t *testing.T)

}

func TestNotifyMsgResendsOnlyChanges(t *testing.T) {
	codec := dataCodec{}

	var cfg KVConfig
	flagext.DefaultValues(&cfg)
	cfg.TCPTransport.BindAddrs = getLocalhostAddrs()
	// We will be checking for number of messages in the broadcast queue, so make sure to use known retransmit factor.
	cfg.RetransmitMult = 1
	cfg.Codecs = append(cfg.Codecs, codec)

	kv := NewKV(cfg, log.NewNopLogger(), &staticDNSProviderMock{}, prometheus.NewPedanticRegistry())
	require.NoError(t, services.StartAndAwaitRunning(context.Background(), kv))
	defer services.StopAndAwaitTerminated(context.Background(), kv) //nolint:errcheck

	client, err := NewClient(kv, codec)
	require.NoError(t, err)

	// No broadcast messages from KV at the beginning.
	require.Equal(t, 0, len(kv.GetBroadcasts(0, math.MaxInt32)))

	now := time.Now()

	require.NoError(t, client.CAS(context.Background(), key, func(in interface{}) (out interface{}, retry bool, err error) {
		d := getOrCreateData(in)
		d.Members["a"] = member{Timestamp: now.Unix(), State: JOINING}
		d.Members["b"] = member{Timestamp: now.Unix(), State: JOINING}
		return d, true, nil
	}))

	// Check that new instance is broadcasted about just once.
	assert.Equal(t, 1, len(kv.GetBroadcasts(0, math.MaxInt32)))
	require.Equal(t, 0, len(kv.GetBroadcasts(0, math.MaxInt32)))

	kv.NotifyMsg(marshalKeyValuePair(t, key, codec, &data{
		Members: map[string]member{
			"a": {Timestamp: now.Unix() - 5, State: ACTIVE},
			"b": {Timestamp: now.Unix() + 5, State: ACTIVE, Tokens: []uint32{1, 2, 3}},
			"c": {Timestamp: now.Unix(), State: ACTIVE},
		}}))

	// Wait until KV update has been processed.
	time.Sleep(time.Millisecond * 100)

	// Check two things here:
	// 1) state of value in KV store
	// 2) broadcast message only has changed members

	d := getData(t, client, key)
	assert.Equal(t, &data{
		Members: map[string]member{
			"a": {Timestamp: now.Unix(), State: JOINING, Tokens: []uint32{}}, // unchanged, timestamp too old
			"b": {Timestamp: now.Unix() + 5, State: ACTIVE, Tokens: []uint32{1, 2, 3}},
			"c": {Timestamp: now.Unix(), State: ACTIVE, Tokens: []uint32{}},
		}}, d)

	bs := kv.GetBroadcasts(0, math.MaxInt32)
	require.Equal(t, 1, len(bs))