feat(compliance-chains): s3 polling + hash-store #NIT-4269 (#4234)

* feat(compliance-chains): s3 polling + hash-store

# Conflicts:
#	arbnode/node.go

* remove address-checker+add lru-cache

* fix lint and add changelog

* move restrictedaddr service to execution node

* move restrictedaddr service to execution node 2

* apply suggestions and enhnace s3 download to use memory buffer

* Fix address-filter configci

* Remove un-needed md file

* rename addressfilter and change copyright

* update changelog

* enhance PR, fix issues

* update changelog and move service enable check mechanism

* update changelog 2

* apply suggestions

* apply suggestions 2

* apply suggestions 3

* fix lint

* fix ci-test

Author: mahdy-nasr

addressfilter/config.go

@@ -0,0 +1,54 @@
+// Copyright 2026, Offchain Labs, Inc.
+// For license information, see https://github.com/OffchainLabs/nitro/blob/master/LICENSE.md
+
+package addressfilter
+
+import (
+	"errors"
+	"time"
+
+	"github.com/spf13/pflag"
+
+	"github.com/offchainlabs/nitro/util/s3syncer"
+)
+
+type Config struct {
+	Enable       bool            `koanf:"enable"`
+	S3           s3syncer.Config `koanf:"s3"`
+	PollInterval time.Duration   `koanf:"poll-interval"`
+	CacheSize    int             `koanf:"cache-size"`
+}
+
+var DefaultConfig = Config{
+	Enable:       false,
+	PollInterval: 5 * time.Minute,
+	CacheSize:    10000,
+	S3:           s3syncer.DefaultS3Config,
+}
+
+func ConfigAddOptions(prefix string, f *pflag.FlagSet) {
+	f.Bool(prefix+".enable", DefaultConfig.Enable, "enable restricted address synchronization service")
+	s3syncer.ConfigAddOptions(prefix+".s3", f)
+	f.Duration(prefix+".poll-interval", DefaultConfig.PollInterval, "interval between polling S3 for hash list updates")
+	f.Int(prefix+".cache-size", DefaultConfig.CacheSize, "LRU cache size for address lookup results")
+}
+
+func (c *Config) Validate() error {
+	if !c.Enable {
+		return nil
+	}
+
+	if err := c.S3.Validate(); err != nil {
+		return err
+	}
+
+	if c.PollInterval <= 0 {
+		return errors.New("address-filter.poll-interval must be positive")
+	}
+
+	if c.CacheSize <= 0 {
+		return errors.New("address-filter.cache-size must be positive")
+	}
+
+	return nil
+}

addressfilter/hash_store.go

@@ -0,0 +1,116 @@
+// Copyright 2026, Offchain Labs, Inc.
+// For license information, see https://github.com/OffchainLabs/nitro/blob/master/LICENSE.md
+
+package addressfilter
+
+import (
+	"crypto/sha256"
+	"sync/atomic"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/lru"
+)
+
+// hashData holds the immutable hash list data.
+// Once created, this struct is never modified, making it safe for concurrent reads.
+// The cache is included here so it gets swapped atomically with the hash data.
+type hashData struct {
+	salt     []byte
+	hashes   map[common.Hash]struct{}
+	digest   string
+	loadedAt time.Time
+	cache    *lru.Cache[common.Address, bool] // LRU cache for address lookup results
+}
+
+// HashStore provides thread-safe access to restricted address hashes.
+// It uses atomic.Pointer for lock-free reads during updates, implementing
+// a double-buffering strategy where new data is prepared in the background
+// and then atomically swapped in.
+type HashStore struct {
+	data      atomic.Pointer[hashData]
+	cacheSize int
+}
+
+func NewHashStore(cacheSize int) *HashStore {
+	h := &HashStore{
+		cacheSize: cacheSize,
+	}
+	h.data.Store(&hashData{
+		hashes: make(map[common.Hash]struct{}),
+		cache:  lru.NewCache[common.Address, bool](cacheSize),
+	})
+	return h
+}
+
+// Store atomically swaps in a new hash list.
+// This is called after a new hash list has been downloaded and parsed.
+// A new LRU cache is created for the new data, ensuring atomic consistency.
+func (h *HashStore) Store(salt []byte, hashes []common.Hash, digest string) {
+	newData := &hashData{
+		salt:     salt,
+		hashes:   make(map[common.Hash]struct{}, len(hashes)),
+		digest:   digest,
+		loadedAt: time.Now(),
+		cache:    lru.NewCache[common.Address, bool](h.cacheSize),
+	}
+	for _, hash := range hashes {
+		newData.hashes[hash] = struct{}{}
+	}
+	h.data.Store(newData) // Atomic pointer swap
+}
+
+// IsRestricted checks if an address is in the restricted list.
+// Results are cached in the LRU cache for faster subsequent lookups.
+// This method is safe to call concurrently.
+func (h *HashStore) IsRestricted(addr common.Address) bool {
+	data := h.data.Load() // Atomic load - no lock needed
+	if len(data.salt) == 0 {
+		return false // Not initialized
+	}
+
+	// Check cache first (cache is per-data snapshot)
+	if restricted, ok := data.cache.Get(addr); ok {
+		return restricted
+	}
+
+	saltedAddr := make([]byte, len(data.salt)+common.AddressLength)
+	copy(saltedAddr, data.salt)
+	copy(saltedAddr[len(data.salt):], addr.Bytes())
+	saltedHash := sha256.Sum256(saltedAddr)
+
+	_, restricted := data.hashes[saltedHash]
+
+	// Cache the result
+	data.cache.Add(addr, restricted)
+	return restricted
+}
+
+// Digest Return the digest of the current loaded hashstore.
+func (h *HashStore) Digest() string {
+	return h.data.Load().digest
+}
+
+func (h *HashStore) Size() int {
+	return len(h.data.Load().hashes)
+}
+
+func (h *HashStore) LoadedAt() time.Time {
+	return h.data.Load().loadedAt
+}
+
+// Salt returns a copy of the current salt.
+func (h *HashStore) Salt() []byte {
+	data := h.data.Load()
+	if len(data.salt) == 0 {
+		return nil
+	}
+	salt := make([]byte, len(data.salt))
+	copy(salt, data.salt)
+	return salt
+}
+
+// CacheLen returns the current number of entries in the LRU cache.
+func (h *HashStore) CacheLen() int {
+	return h.data.Load().cache.Len()
+}

addressfilter/s3_sync.go

@@ -0,0 +1,97 @@
+// Copyright 2026, Offchain Labs, Inc.
+// For license information, see https://github.com/OffchainLabs/nitro/blob/master/LICENSE.md
+
+package addressfilter
+
+import (
+	"context"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/log"
+
+	"github.com/offchainlabs/nitro/util/s3syncer"
+)
+
+// hashListPayload represents the JSON structure of the hash list file used for unmarshalling.
+type hashListPayload struct {
+	Salt          string `json:"salt"`
+	HashingScheme string `json:"hashing_scheme,omitempty"`
+	AddressHashes []struct {
+		Hash string `json:"hash"`
+	} `json:"address_hashes"`
+}
+
+type S3SyncManager struct {
+	Syncer    *s3syncer.Syncer
+	hashStore *HashStore
+}
+
+func NewS3SyncManager(ctx context.Context, config *Config, hashStore *HashStore) (*S3SyncManager, error) {
+	s := &S3SyncManager{
+		hashStore: hashStore,
+	}
+	syncer, err := s3syncer.NewSyncer(
+		ctx,
+		&config.S3,
+		s.handleHashListData,
+	)
+
+	if err != nil {
+		return nil, err
+	}
+
+	s.Syncer = syncer
+	return s, nil
+}
+
+// handleHashListData parses the downloaded JSON data and loads it into the hashStore.
+func (s *S3SyncManager) handleHashListData(data []byte, digest string) error {
+	salt, hashes, err := parseHashListJSON(data)
+	if err != nil {
+		return fmt.Errorf("failed to parse hash list: %w", err)
+	}
+
+	s.hashStore.Store(salt, hashes, digest)
+	log.Info("loaded restricted addr list", "hash_count", len(hashes), "etag", digest, "size_bytes", len(data))
+	return nil
+}
+
+// parseHashListJSON parses the JSON hash list file.
+// Expected format: {"salt": "hex...", "address_hashes": [{"hash": "hex1"}, {"hash": "hex2"}, ...]}
+func parseHashListJSON(data []byte) ([]byte, []common.Hash, error) {
+	var payload hashListPayload
+	if err := json.Unmarshal(data, &payload); err != nil {
+		return nil, nil, fmt.Errorf("JSON unmarshal failed: %w", err)
+	}
+
+	// Validate hashing scheme - warn if not Sha256 but continue for forward compatibility
+	if payload.HashingScheme != "" && payload.HashingScheme != "Sha256" {
+		log.Warn("unknown hashing scheme in address list, continuing with Sha256 assumption",
+			"scheme", payload.HashingScheme)
+	}
+
+	salt, err := hex.DecodeString(payload.Salt)
+	if err != nil {
+		return nil, nil, fmt.Errorf("invalid salt hex: %w", err)
+	}
+	if len(salt) == 0 {
+		return nil, nil, fmt.Errorf("salt cannot be empty")
+	}
+
+	hashes := make([]common.Hash, len(payload.AddressHashes))
+	for i, h := range payload.AddressHashes {
+		hashBytes, err := hex.DecodeString(h.Hash)
+		if err != nil {
+			return nil, nil, fmt.Errorf("invalid hash hex at index %d: %w", i, err)
+		}
+		if len(hashBytes) != 32 {
+			return nil, nil, fmt.Errorf("invalid hash length at index %d: got %d, want 32", i, len(hashBytes))
+		}
+		copy(hashes[i][:], hashBytes)
+	}
+
+	return salt, hashes, nil
+}

addressfilter/service.go

@@ -0,0 +1,116 @@
+// Copyright 2026, Offchain Labs, Inc.
+// For license information, see https://github.com/OffchainLabs/nitro/blob/master/LICENSE.md
+
+package addressfilter
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/ethereum/go-ethereum/log"
+
+	"github.com/offchainlabs/nitro/util/stopwaiter"
+)
+
+// Service manages the address-filteress synchronization pipeline.
+// It periodically polls S3 for hash list updates and maintains an in-memory
+// copy for efficient address filtering.
+type FilterService struct {
+	stopwaiter.StopWaiter
+	config    *Config
+	hashStore *HashStore
+	syncMgr   *S3SyncManager
+}
+
+// NewFilterService creates a new address-filteress service.
+// Returns nil if the service is not enabled in the configuration.
+func NewFilterService(ctx context.Context, config *Config) (*FilterService, error) {
+	if !config.Enable {
+		return nil, nil
+	}
+
+	if err := config.Validate(); err != nil {
+		return nil, fmt.Errorf("invalid config: %w", err)
+	}
+
+	hashStore := NewHashStore(config.CacheSize)
+	syncMgr, err := NewS3SyncManager(ctx, config, hashStore)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create S3 syncer: %w", err)
+	}
+
+	return &FilterService{
+		config:    config,
+		hashStore: hashStore,
+		syncMgr:   syncMgr,
+	}, nil
+}
+
+// Initialize downloads the initial hash list from S3.
+// This method blocks until the hash list is successfully loaded.
+// If this fails, the node should not start.
+func (s *FilterService) Initialize(ctx context.Context) error {
+	log.Info("initializing address-filter service, downloading initial hash list",
+		"bucket", s.config.S3.Bucket,
+		"key", s.config.S3.ObjectKey,
+	)
+
+	// Force download (ignore ETag check for initial load)
+	if err := s.syncMgr.Syncer.DownloadAndLoad(ctx); err != nil {
+		return fmt.Errorf("failed to load initial hash list: %w", err)
+	}
+
+	log.Info("address-filter service initialized",
+		"hash_count", s.hashStore.Size(),
+		"etag-digest", s.hashStore.Digest(),
+	)
+	return nil
+}
+
+// Start begins the background polling goroutine.
+// This should be called after Initialize() succeeds.
+func (s *FilterService) Start(ctx context.Context) {
+	s.StopWaiter.Start(ctx, s)
+
+	// Start periodic polling goroutine
+	s.CallIteratively(func(ctx context.Context) time.Duration {
+		if err := s.syncMgr.Syncer.CheckAndSync(ctx); err != nil {
+			log.Error("failed to sync address-filter list", "err", err)
+		}
+		return s.config.PollInterval
+	})
+
+	log.Info("address-filter service started",
+		"poll_interval", s.config.PollInterval,
+	)
+}
+
+func (s *FilterService) GetHashCount() int {
+	if !s.config.Enable {
+		return 0
+	}
+	return s.hashStore.Size()
+}
+
+// GetHashStoreDigest GetETag returns the S3 ETag Digest of the currently loaded hash list.
+func (s *FilterService) GetHashStoreDigest() string {
+	if !s.config.Enable {
+		return ""
+	}
+	return s.hashStore.Digest()
+}
+
+func (s *FilterService) GetLoadedAt() time.Time {
+	if !s.config.Enable {
+		return time.Time{}
+	}
+	return s.hashStore.LoadedAt()
+}
+
+func (s *FilterService) GetHashStore() *HashStore {
+	if !s.config.Enable {
+		return nil
+	}
+	return s.hashStore
+}