apply suggestions and enhnace s3 download to use memory buffer

Author: mahdy-nasr

address-filter/ARCHITECTURE.md

@@ -0,0 +1,148 @@
+# Restricted Address Filtering - Architecture
+
+## Overview
+
+The `restrictedaddr` package provides compliance-based address filtering for Nitro sequencers. It maintains a list of restricted address hashes (loaded from S3) and blocks transactions involving those addresses.
+
+## Architecture Diagram
+
+```mermaid
+flowchart LR
+    subgraph DataSync["Data Synchronization"]
+        direction TB
+        S3[("S3 Bucket")]
+        S3Sync["S3Syncer"]
+        S3 -->|"HeadObject<br/>(ETag check)"| S3Sync
+        S3 -->|"GetObject<br/>(if changed)"| S3Sync
+    end
+
+    subgraph Storage["In-Memory Storage"]
+        direction TB
+        HashStore["HashStore<br/>atomic.Pointer"]
+        LRU["LRU Cache<br/>(10k entries)"]
+        HashStore --> LRU
+    end
+
+    subgraph TxProcessing["Transaction Processing"]
+        direction TB
+        UserTx(("User Tx"))
+        ExecEngine["ExecutionEngine"]
+        TxFilter["txfilter"]
+        Sequencer["Sequencer"]
+
+        UserTx --> ExecEngine
+        ExecEngine --> TxFilter
+        TxFilter -->|"allowed"| Sequencer
+    end
+
+    S3Sync -->|"atomic swap"| HashStore
+    TxFilter -.->|"IsRestricted?"| HashStore
+
+    style S3 fill:#f5f5f5,stroke:#424242,stroke-width:2px,color:#212121
+    style S3Sync fill:#fff,stroke:#424242,stroke-width:1px,color:#212121
+    style HashStore fill:#e3f2fd,stroke:#1565c0,stroke-width:2px,color:#0d47a1
+    style LRU fill:#e3f2fd,stroke:#1565c0,stroke-width:1px,color:#0d47a1
+    style UserTx fill:#fff,stroke:#424242,stroke-width:1px,color:#212121
+    style ExecEngine fill:#f5f5f5,stroke:#424242,stroke-width:1px,color:#212121
+    style TxFilter fill:#fff3e0,stroke:#e65100,stroke-width:2px,color:#bf360c
+    style Sequencer fill:#f5f5f5,stroke:#424242,stroke-width:1px,color:#212121
+
+    style DataSync fill:#fafafa,stroke:#9e9e9e,stroke-width:1px,color:#212121
+    style Storage fill:#e3f2fd,stroke:#1565c0,stroke-width:1px,color:#0d47a1
+    style TxProcessing fill:#fafafa,stroke:#9e9e9e,stroke-width:1px,color:#212121
+```
+
+## Data Flow
+
+| Flow | Description |
+|------|-------------|
+| **Sync** | S3 → S3Syncer → HashStore (atomic swap on ETag change) |
+| **Lookup** | txfilter → HashStore → LRU cache or `sha256(salt \|\| addr)` hash lookup |
+| **Transaction** | User → ExecutionEngine → txfilter → Sequencer |
+
+## Components
+
+| Component | Role |
+|-----------|------|
+| **Service** | Orchestrates lifecycle: initialization, polling, shutdown |
+| **S3Syncer** | Polls S3 via HeadObject, downloads on ETag change (multipart, 10 concurrent parts) |
+| **HashStore** | Lock-free storage with `atomic.Pointer`, per-snapshot LRU cache (10k entries) |
+| **txfilter** | Blocks transactions touching restricted addresses |
+
+## Package Structure
+
+```
+restrictedaddr/
+├── config.go       # Configuration struct and validation
+├── service.go      # Service lifecycle (Initialize, Start, Stop)
+├── s3_sync.go      # S3 polling and concurrent download
+├── hash_store.go   # Lock-free hash storage with LRU caching
+└── service_test.go # Unit and integration tests
+```
+
+## Configuration
+
+| Option | Default | Description |
+|--------|---------|-------------|
+| `--restricted-addr.enable` | `false` | Enable the service |
+| `--restricted-addr.s3-bucket` | - | S3 bucket name (required if enabled) |
+| `--restricted-addr.s3-region` | - | AWS region (required if enabled) |
+| `--restricted-addr.s3-object-key` | - | Path to hash list JSON (required if enabled) |
+| `--restricted-addr.s3-access-key` | - | AWS access key (optional, uses default chain) |
+| `--restricted-addr.s3-secret-key` | - | AWS secret key (optional) |
+| `--restricted-addr.poll-interval` | `5m` | Interval between S3 ETag checks |
+
+## S3 Hash List Format
+
+```json
+{
+  "salt": "hex_encoded_salt",
+  "address_hashes": [
+    {"hash": "hex_encoded_32byte_sha256"},
+    {"hash": "hex_encoded_32byte_sha256"}
+  ]
+}
+```
+
+**Hash computation:** `SHA256(salt || address_bytes)` - addresses are never stored in plaintext.
+
+## Service Lifecycle
+
+1. **NewService** - Validates config, creates S3 client
+2. **Initialize** - Blocking initial download (node won't start if this fails)
+3. **Start** - Begins background polling goroutine
+4. **StopAndWait** - Graceful shutdown
+
+The service initializes **early** in the node startup sequence (before inbox tracker, transaction streamer, etc.) to ensure filtering is active before any transactions are processed.
+
+## HashStore Design
+
+- **Lock-free reads:** Uses `atomic.Pointer[hashData]` for concurrent access
+- **Double-buffering:** New data prepared while old data still serves requests
+- **Per-snapshot LRU:** Each atomic swap includes a fresh 10k-entry cache
+- **Lookup methods:**
+  - `IsRestricted(addr)` - Single address check
+  - `IsAnyRestricted(addrs)` - True if any address restricted
+  - `IsAllRestricted(addrs)` - True only if all addresses restricted
+
+## Transaction Filtering Points
+
+Transactions are blocked if any of these addresses are restricted:
+
+| Operation | Addresses Checked |
+|-----------|-------------------|
+| Transfer | Sender, Recipient |
+| CALL | Target contract |
+| STATICCALL | Target contract |
+| CREATE/CREATE2 | New contract address |
+| SELFDESTRUCT | Beneficiary address |
+
+## S3 Polling Strategy
+
+1. **HeadObject** call to check ETag (lightweight, no data transfer)
+2. If ETag unchanged, skip download
+3. If ETag changed:
+   - Download to temp file (multipart: 32MB parts, 10 concurrent, 5 retries/part)
+   - Parse and validate JSON
+   - Atomic pointer swap into HashStore
+4. Repeat after `poll-interval`

address-filter/config.go

@@ -0,0 +1,46 @@
+// Copyright 2025, 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"`
+}
+
+var DefaultConfig = Config{
+	Enable:       false,
+	PollInterval: 5 * time.Minute,
+}
+
+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")
+}
+
+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("restricted-addr.poll-interval must be positive")
+	}
+
+	return nil
+}

restrictedaddr/hash_store.go address-filter/hash_store.gorestrictedaddr/hash_store.go renamed to address-filter/hash_store.go

@@ -1,7 +1,7 @@
 // Copyright 2025, Offchain Labs, Inc.
 // For license information, see https://github.com/OffchainLabs/nitro/blob/master/LICENSE.md
 
-package restrictedaddr
+package addressfilter
 
 import (
 	"crypto/sha256"
@@ -17,7 +17,7 @@ import (
 // The cache is included here so it gets swapped atomically with the hash data.
 type hashData struct {
 	salt     []byte
-	hashes   map[[32]byte]struct{}
+	hashes   map[common.Hash]struct{}
 	digest   string
 	loadedAt time.Time
 	cache    *lru.Cache[common.Address, bool] // LRU cache for address lookup results
@@ -43,7 +43,7 @@ func NewHashStoreWithCacheSize(cacheSize int) *HashStore {
 		cacheSize: cacheSize,
 	}
 	h.data.Store(&hashData{
-		hashes: make(map[[32]byte]struct{}),
+		hashes: make(map[common.Hash]struct{}),
 		cache:  lru.NewCache[common.Address, bool](cacheSize),
 	})
 	return h
@@ -52,10 +52,10 @@ func NewHashStoreWithCacheSize(cacheSize int) *HashStore {
 // Load 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) Load(salt []byte, hashes [][32]byte, digest string) {
+func (h *HashStore) Load(salt []byte, hashes []common.Hash, digest string) {
 	newData := &hashData{
 		salt:     salt,
-		hashes:   make(map[[32]byte]struct{}, len(hashes)),
+		hashes:   make(map[common.Hash]struct{}, len(hashes)),
 		digest:   digest,
 		loadedAt: time.Now(),
 		cache:    lru.NewCache[common.Address, bool](h.cacheSize),
@@ -88,58 +88,6 @@ func (h *HashStore) IsRestricted(addr common.Address) bool {
 	return restricted
 }
 
-// IsAllRestricted checks if all provided addresses are in the restricted list
-// from same hash-store snapshot. Results are cached in the LRU cache.
-func (h *HashStore) IsAllRestricted(addrs []common.Address) bool {
-	data := h.data.Load() // Atomic load - no lock needed
-	if len(data.salt) == 0 {
-		return false // Not initialized
-	}
-	for _, addr := range addrs {
-		// Check cache first (cache is per-data snapshot)
-		if restricted, ok := data.cache.Get(addr); ok {
-			if !restricted {
-				return false
-			}
-			continue
-		}
-
-		hash := sha256.Sum256(append(data.salt, addr.Bytes()...))
-		_, restricted := data.hashes[hash]
-		data.cache.Add(addr, restricted)
-		if !restricted {
-			return false
-		}
-	}
-	return true
-}
-
-// IsAnyRestricted checks if any of the provided addresses are in the restricted list
-// from same hash-store snapshot. Results are cached in the LRU cache.
-func (h *HashStore) IsAnyRestricted(addrs []common.Address) bool {
-	data := h.data.Load() // Atomic load - no lock needed
-	if len(data.salt) == 0 {
-		return false // Not initialized
-	}
-	for _, addr := range addrs {
-		// Check cache first (cache is per-data snapshot)
-		if restricted, ok := data.cache.Get(addr); ok {
-			if restricted {
-				return true
-			}
-			continue
-		}
-
-		hash := sha256.Sum256(append(data.salt, addr.Bytes()...))
-		_, restricted := data.hashes[hash]
-		data.cache.Add(addr, restricted)
-		if restricted {
-			return true
-		}
-	}
-	return false
-}
-
 // Digest Return the digest of the current loaded hashstore.
 func (h *HashStore) Digest() string {
 	return h.data.Load().digest

address-filter/s3_sync.go

@@ -0,0 +1,92 @@
+// Copyright 2025, 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"`
+	AddressHashes []struct {
+		Hash string `json:"hash"`
+	} `json:"address_hashes"`
+}
+
+type S3SyncManager struct {
+	Syncer *s3syncer.Syncer
+	store  *HashStore
+}
+
+func NewS3SyncManager(ctx context.Context, config *Config, store *HashStore) (*S3SyncManager, error) {
+	s := &S3SyncManager{
+		store: store,
+	}
+	syncer, err := s3syncer.NewSyncer(
+		ctx,
+		&config.S3,
+		s.handleHashListData,
+		// These are initial settings that can be tuned as needed.
+		s3syncer.WithDownloadConfig(s3syncer.DownloadConfig{
+			PartSizeMB:         100,
+			Concurrency:        10,
+			PartBodyMaxRetries: 5,
+		}))
+
+	if err != nil {
+		return nil, err
+	}
+
+	s.Syncer = syncer
+	return s, nil
+}
+
+// handleHashListData parses the downloaded JSON data and loads it into the store.
+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.store.Load(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)
+	}
+
+	salt, err := hex.DecodeString(payload.Salt)
+	if err != nil {
+		return nil, nil, fmt.Errorf("invalid salt hex: %w", err)
+	}
+
+	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
+}