BCFR-1201 bind lp filter fix race

.changeset/funny-wombats-teach.md

@@ -0,0 +1,5 @@
+---
+"chainlink": patch
+---
+
+#bugfix bind reg filter before unreg & bind noop when no new addresses

core/services/relay/evm/read/bindings_test.go

@@ -112,4 +112,45 @@ func TestBindingsRegistry(t *testing.T) {
 		mRdr1.AssertExpectations(t)
 		mReg.AssertExpectations(t)
 	})
+
+	t.Run("bind twice is noop", func(t *testing.T) {
+		t.Parallel()
+
+		mRdr0 := new(mocks.Reader)
+		mReg := new(mocks.Registrar)
+
+		named := read.NewBindingsRegistry()
+
+		// register is called once through RegisterAll and again in Bind
+		mRdr0.EXPECT().Register(mock.Anything).Return(nil)
+
+		mRdr0.EXPECT().Bind(mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+
+		mReg.EXPECT().HasFilter(mock.Anything).Return(false)
+		mReg.EXPECT().RegisterFilter(mock.Anything, mock.Anything).Return(nil).Times(4) //2 times per init + 1 time per bind(3) + 1 time per bind new address
+		// part of the init phase of chain reader
+		require.NoError(t, named.AddReader(contractName1, methodName1, mRdr0))
+		_ = named.SetFilter(contractName1, filterWithSigs)
+
+		// run within the start phase of chain reader
+		require.NoError(t, named.RegisterAll(context.Background(), mReg))
+
+		bindings := []commontypes.BoundContract{
+			{Address: "0x24", Name: contractName1},
+			{Address: "0x25", Name: contractName1},
+			{Address: "0x26", Name: contractName1},
+		}
+
+		// calling bind now should trigger register filter just once
+		_ = named.Bind(context.Background(), mReg, bindings)
+		_ = named.Bind(context.Background(), mReg, bindings)
+		_ = named.Bind(context.Background(), mReg, bindings)
+
+		// this will trigger register filter
+		bindings[0].Address = "0x99"
+		_ = named.Bind(context.Background(), mReg, bindings)
+
+		mRdr0.AssertExpectations(t)
+		mReg.AssertExpectations(t)
+	})
 }

core/services/relay/evm/read/contract.go

@@ -60,28 +60,20 @@ func (cb *contractBinding) AddReaderNamed(name string, rdr Reader) {
 
 // Bind binds contract addresses to contract binding and registers the common contract polling filter.
 func (cb *contractBinding) Bind(ctx context.Context, registrar Registrar, bindings ...common.Address) error {
-	if cb.isBound() {
-		if err := cb.Unregister(ctx, registrar); err != nil {
-			return err
-		}
-	}
-
 	for _, binding := range bindings {
 		if cb.bindingExists(binding) {
 			continue
 		}
 
-		cb.registrar.SetName(logpoller.FilterName(cb.name + "." + uuid.NewString()))
 		cb.registrar.AddAddress(binding)
 		cb.addBinding(binding)
 	}
 
-	// registerCalled during ChainReader start
-	if cb.registered() {
-		return cb.Register(ctx, registrar)
+	if !cb.registrar.Dirty() {
+		return nil
 	}
 
-	return nil
+	return cb.Update(ctx, registrar)
 }
 
 func (cb *contractBinding) BindReaders(ctx context.Context, addresses ...common.Address) error {
@@ -159,6 +151,21 @@ func (cb *contractBinding) Register(ctx context.Context, registrar Registrar) er
 	return nil
 }
 
+func (cb *contractBinding) Update(ctx context.Context, registrar Registrar) error {
+	name := logpoller.FilterName(cb.name + "." + uuid.NewString())
+
+	if !cb.registered() {
+		cb.registrar.SetName(name)
+		return nil
+	}
+
+	if !cb.registrar.HasEventSigs() {
+		return nil
+	}
+
+	return cb.registrar.Update(ctx, registrar, name)
+}
+
 func (cb *contractBinding) RegisterReaders(ctx context.Context) error {
 	cb.mu.RLock()
 	defer cb.mu.RUnlock()

core/services/relay/evm/read/event.go

@@ -134,18 +134,6 @@ func (b *EventBinding) GetDataWords() map[string]DataWordDetail {
 }
 
 func (b *EventBinding) Bind(ctx context.Context, bindings ...common.Address) error {
-	if b.hasBindings() {
-		// we are changing contract address reference, so we need to unregister old filter if it exists
-		if err := b.Unregister(ctx); err != nil {
-			return err
-		}
-	}
-
-	// filterRegisterer isn't required here because the event can also be polled for by the contractBinding common filter.
-	if b.registrar != nil {
-		b.registrar.SetName(logpoller.FilterName(fmt.Sprintf("%s.%s.%s", b.contractName, b.eventName, uuid.NewString())))
-	}
-
 	for _, binding := range bindings {
 		if b.isBound(binding) {
 			continue
@@ -156,10 +144,15 @@ func (b *EventBinding) Bind(ctx context.Context, bindings ...common.Address) err
 		}
 
 		b.addBinding(binding)
+
+	}
+
+	if b.registrar == nil || !b.registrar.Dirty() {
+		return nil
 	}
 
 	if b.registered() {
-		return b.Register(ctx)
+		return b.Update(ctx)
 	}
 
 	return nil
@@ -196,6 +189,23 @@ func (b *EventBinding) Unbind(ctx context.Context, bindings ...common.Address) e
 	return nil
 }
 
+func (b *EventBinding) Update(ctx context.Context) error {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+
+	name := logpoller.FilterName(fmt.Sprintf("%s.%s.%s", b.contractName, b.eventName, uuid.NewString()))
+
+	if b.registrar == nil {
+		return nil
+	}
+
+	if len(b.bound) == 0 {
+		return nil
+	}
+
+	return b.registrar.Update(ctx, b.lp, name)
+}
+
 func (b *EventBinding) Register(ctx context.Context) error {
 	b.mu.Lock()
 	defer b.mu.Unlock()

core/services/relay/evm/read/filter.go

@@ -22,16 +22,44 @@ type syncedFilter struct {
 	// internal state properties
 	mu     sync.RWMutex
 	filter logpoller.Filter
+
+	// identifies if filter was modified between updates
+	dirty bool
 }
 
 func newSyncedFilter() *syncedFilter {
 	return &syncedFilter{}
 }
 
+func (r *syncedFilter) Update(ctx context.Context, registrar Registrar, updatedName string) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	oldName := r.filter.Name
+	if !r.dirty {
+		return nil
+	}
+
+	r.filter.Name = updatedName
+
+	if err := r.register(ctx, registrar); err != nil {
+		return err
+	}
+
+	// filter updated successfully, it's not dirty anymore
+	r.dirty = false
+
+	return r.unregister(ctx, registrar, oldName)
+}
+
 func (r *syncedFilter) Register(ctx context.Context, registrar Registrar) error {
 	r.mu.RLock()
 	defer r.mu.RUnlock()
 
+	return r.register(ctx, registrar)
+}
+
+func (r *syncedFilter) register(ctx context.Context, registrar Registrar) error {
 	if !registrar.HasFilter(r.filter.Name) {
 		if err := registrar.RegisterFilter(ctx, r.filter); err != nil {
 			return FilterError{
@@ -49,11 +77,15 @@ func (r *syncedFilter) Unregister(ctx context.Context, registrar Registrar) erro
 	r.mu.RLock()
 	defer r.mu.RUnlock()
 
-	if !registrar.HasFilter(r.filter.Name) {
+	return r.unregister(ctx, registrar, r.filter.Name)
+}
+
+func (r *syncedFilter) unregister(ctx context.Context, registrar Registrar, name string) error {
+	if !registrar.HasFilter(name) {
 		return nil
 	}
 
-	if err := registrar.UnregisterFilter(ctx, r.filter.Name); err != nil {
+	if err := registrar.UnregisterFilter(ctx, name); err != nil {
 		return FilterError{
 			Err:    fmt.Errorf("%w: %s", types.ErrInternal, err.Error()),
 			Action: "unregister",
@@ -68,27 +100,35 @@ func (r *syncedFilter) SetFilter(filter logpoller.Filter) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
+	r.dirty = true
+
 	r.filter = filter
 }
 
 func (r *syncedFilter) SetName(name string) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
+	r.dirty = true
+
 	r.filter.Name = name
 }
 
 func (r *syncedFilter) AddAddress(address common.Address) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
+	r.dirty = true
+
 	r.filter.Addresses = append(r.filter.Addresses, address)
 }
 
 func (r *syncedFilter) RemoveAddress(address common.Address) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
+	r.dirty = true
+
 	var addrIdx int
 	for idx, addr := range r.filter.Addresses {
 		if addr.Hex() == address.Hex() {
@@ -107,6 +147,13 @@ func (r *syncedFilter) Count() int {
 	return len(r.filter.Addresses)
 }
 
+func (r *syncedFilter) Dirty() bool {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+
+	return r.dirty
+}
+
 func (r *syncedFilter) HasEventSigs() bool {
 	r.mu.RLock()
 	defer r.mu.RUnlock()

core/services/relay/evm/read/lookup.go

@@ -34,12 +34,7 @@ func (l *lookup) addReadNameForContract(contract, readName string) {
 	l.mu.Lock()
 	defer l.mu.Unlock()
 
-	readNames, exists := l.contractReadNames[contract]
-	if !exists {
-		readNames = []string{}
-	}
-
-	l.contractReadNames[contract] = append(readNames, readName)
+	l.contractReadNames[contract] = append(l.contractReadNames[contract], readName)
 }
 
 func (l *lookup) bindAddressForContract(contract, address string) {