NONEVM-747 register unregister filters

.mockery.yaml

@@ -40,3 +40,9 @@ packages:
   github.com/smartcontractkit/chainlink-solana/pkg/solana/logpoller:
     interfaces:
       RPCClient:
+      ORM:
+        config:
+          inpackage: True
+          dir: "pkg/solana/logpoller"
+          filename: mock_orm.go
+          mockname: mockORM

go.mod

@@ -1,6 +1,6 @@
 module github.com/smartcontractkit/chainlink-solana
 
-go 1.22.0
+go 1.23
 
 toolchain go1.23.1
 

integration-tests/go.mod

@@ -1,6 +1,6 @@
 module github.com/smartcontractkit/chainlink-solana/integration-tests
 
-go 1.22.8
+go 1.23
 
 toolchain go1.23.2
 

pkg/solana/logpoller/filters.go

@@ -0,0 +1,262 @@
+package logpoller
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"iter"
+	"maps"
+	"sync"
+	"sync/atomic"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+)
+
+type filters struct {
+	orm  ORM
+	lggr logger.SugaredLogger
+
+	filtersByName     map[string]Filter
+	filtersByAddress  map[PublicKey]map[EventSignature]map[int64]Filter
+	filtersToBackfill map[int64]Filter
+	filtersToDelete   map[int64]Filter
+	filtersMutex      sync.RWMutex
+	loadedFilters     atomic.Bool
+}
+
+func newFilters(lggr logger.SugaredLogger, orm ORM) *filters {
+	return &filters{
+		orm:  orm,
+		lggr: lggr,
+	}
+}
+
+// PruneFilters - prunes all filters marked to be deleted from the database and all corresponding logs.
+func (fl *filters) PruneFilters(ctx context.Context) error {
+	err := fl.LoadFilters(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to load filters: %w", err)
+	}
+
+	fl.filtersMutex.Lock()
+	filtersToDelete := fl.filtersToDelete
+	fl.filtersToDelete = make(map[int64]Filter)
+	fl.filtersMutex.Unlock()
+
+	if len(filtersToDelete) == 0 {
+		return nil
+	}
+
+	err = fl.orm.DeleteFilters(ctx, filtersToDelete)
+	if err != nil {
+		fl.filtersMutex.Lock()
+		defer fl.filtersMutex.Unlock()
+		maps.Copy(fl.filtersToDelete, filtersToDelete)
+		return fmt.Errorf("failed to delete filters: %w", err)
+	}
+
+	return nil
+}
+
+// RegisterFilter persists provided filter and ensures that any log emitted by a contract with filter.Address
+// that matches filter.EventSig signature will be captured starting from filter.StartingBlock.
+// The filter may be unregistered later by filter.Name.
+// In case of Filter.Name collision (within the chain scope) returns ErrFilterNameConflict if
+// one of the fields defining resulting logs (Address, EventSig, EventIDL, SubkeyPaths) does not match original filter.
+// Otherwise, updates remaining fields and schedules backfill.
+// Warnings/debug information is keyed by filter name.
+func (fl *filters) RegisterFilter(ctx context.Context, filter Filter) error {
+	if len(filter.Name) == 0 {
+		return errors.New("name is required")
+	}
+
+	err := fl.LoadFilters(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to load filters: %w", err)
+	}
+
+	fl.filtersMutex.Lock()
+	defer fl.filtersMutex.Unlock()
+
+	if existingFilter, ok := fl.filtersByName[filter.Name]; ok {
+		if !existingFilter.MatchSameLogs(filter) {
+			return ErrFilterNameConflict
+		}
+
+		fl.removeFilterFromIndexes(existingFilter)
+	}
+
+	filterID, err := fl.orm.InsertFilter(ctx, filter)
+	if err != nil {
+		return fmt.Errorf("failed to insert filter: %w", err)
+	}
+
+	filter.ID = filterID
+	fl.filtersByName[filter.Name] = filter
+	filtersByEventSig, ok := fl.filtersByAddress[filter.Address]
+	if !ok {
+		filtersByEventSig = make(map[EventSignature]map[int64]Filter)
+		fl.filtersByAddress[filter.Address] = filtersByEventSig
+	}
+
+	filtersByID, ok := filtersByEventSig[filter.EventSig]
+	if !ok {
+		filtersByID = make(map[int64]Filter)
+		filtersByEventSig[filter.EventSig] = filtersByID
+	}
+
+	filtersByID[filter.ID] = filter
+	fl.filtersToBackfill[filterID] = filter
+	return nil
+}
+
+// UnregisterFilter will mark the filter with the given name for pruning and async prune all corresponding logs.
+// If the name does not exist, it will log an error but not return an error.
+// Warnings/debug information is keyed by filter name.
+func (fl *filters) UnregisterFilter(ctx context.Context, name string) error {
+	err := fl.LoadFilters(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to load filters: %w", err)
+	}
+
+	fl.filtersMutex.Lock()
+	defer fl.filtersMutex.Unlock()
+
+	filter, ok := fl.filtersByName[name]
+	if !ok {
+		fl.lggr.Warnw("Filter not found in filtersByName", "name", name)
+		return nil
+	}
+
+	if err := fl.orm.MarkFilterDeleted(ctx, filter.ID); err != nil {
+		return fmt.Errorf("failed to mark filter deleted: %w", err)
+	}
+
+	fl.removeFilterFromIndexes(filter)
+
+	fl.filtersToDelete[filter.ID] = filter
+	return nil
+}
+
+func (fl *filters) removeFilterFromIndexes(filter Filter) {
+	delete(fl.filtersByName, filter.Name)
+	delete(fl.filtersToBackfill, filter.ID)
+
+	filtersByEventSig, ok := fl.filtersByAddress[filter.Address]
+	if !ok {
+		fl.lggr.Warnw("Filter not found in filtersByAddress", "name", filter.Name, "address", filter.Address)
+		return
+	}
+
+	filtersByID, ok := filtersByEventSig[filter.EventSig]
+	if !ok {
+		fl.lggr.Warnw("Filter not found in filtersByEventSig", "name", filter.Name, "address", filter.Address)
+		return
+	}
+
+	delete(filtersByID, filter.ID)
+	if len(filtersByID) == 0 {
+		delete(filtersByEventSig, filter.EventSig)
+	}
+
+	if len(filtersByEventSig) == 0 {
+		delete(fl.filtersByAddress, filter.Address)
+	}
+}
+
+// MatchingFilters - returns iterator to go through all matching filters.
+// Requires LoadFilters to be called at least once.
+func (fl *filters) MatchingFilters(addr PublicKey, eventSignature EventSignature) iter.Seq[Filter] {
+	if !fl.loadedFilters.Load() {
+		fl.lggr.Critical("Invariant violation: expected filters to be loaded before call to MatchingFilters")
+		return nil
+	}
+	return func(yield func(Filter) bool) {
+		fl.filtersMutex.RLock()
+		defer fl.filtersMutex.RUnlock()
+		filters, ok := fl.filtersByAddress[addr]
+		if !ok {
+			return
+		}
+
+		for _, filter := range filters[eventSignature] {
+			if !yield(filter) {
+				return
+			}
+		}
+	}
+}
+
+// ConsumeFiltersToBackfill - removes all filters from the backfill queue and returns them to caller.
+// Requires LoadFilters to be called at least once.
+func (fl *filters) ConsumeFiltersToBackfill() map[int64]Filter {
+	if !fl.loadedFilters.Load() {
+		fl.lggr.Critical("Invariant violation: expected filters to be loaded before call to MatchingFilters")
+		return nil
+	}
+	fl.filtersMutex.Lock()
+	defer fl.filtersMutex.Unlock()
+	filtersToBackfill := fl.filtersToBackfill
+	fl.filtersToBackfill = make(map[int64]Filter)
+	return filtersToBackfill
+}
+
+// LoadFilters - loads filters from database. Can be called multiple times without side effects.
+func (fl *filters) LoadFilters(ctx context.Context) error {
+	if fl.loadedFilters.Load() {
+		return nil
+	}
+
+	fl.lggr.Debugw("Loading filters from db")
+	fl.filtersMutex.Lock()
+	defer fl.filtersMutex.Unlock()
+	// reset filters' indexes to ensure we do not have partial data from the previous run
+	fl.filtersByName = make(map[string]Filter)
+	fl.filtersByAddress = make(map[PublicKey]map[EventSignature]map[int64]Filter)
+	fl.filtersToBackfill = make(map[int64]Filter)
+	fl.filtersToDelete = make(map[int64]Filter)
+
+	filters, err := fl.orm.SelectFilters(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to select filters from db: %w", err)
+	}
+
+	for _, filter := range filters {
+		if filter.IsDeleted {
+			fl.filtersToDelete[filter.ID] = filter
+			continue
+		}
+
+		if _, ok := fl.filtersByName[filter.Name]; ok {
+			errMsg := fmt.Sprintf("invariant violation while loading from db: expected filters to have unique name: %s ", filter.Name)
+			fl.lggr.Critical(errMsg)
+			return errors.New(errMsg)
+		}
+
+		fl.filtersByName[filter.Name] = filter
+		filtersByEventSig, ok := fl.filtersByAddress[filter.Address]
+		if !ok {
+			filtersByEventSig = make(map[EventSignature]map[int64]Filter)
+			fl.filtersByAddress[filter.Address] = filtersByEventSig
+		}
+
+		filtersByID, ok := filtersByEventSig[filter.EventSig]
+		if !ok {
+			filtersByID = make(map[int64]Filter)
+			filtersByEventSig[filter.EventSig] = filtersByID
+		}
+
+		if _, ok := filtersByID[filter.ID]; ok {
+			errMsg := fmt.Sprintf("invariant violation while loading from db: expected filters to have unique ID: %d ", filter.ID)
+			fl.lggr.Critical(errMsg)
+			return errors.New(errMsg)
+		}
+
+		filtersByID[filter.ID] = filter
+		fl.filtersToBackfill[filter.ID] = filter
+	}
+
+	fl.loadedFilters.Store(true)
+
+	return nil
+}