evm-gateway-demo: multi-replica demo infra, namespace filtering, block DB retention

- Add namespace filtering to Chain (NewChain gains namespace param)
- Add SQLite PRAGMA temp_store=MEMORY to fix crash on scratch images
- Add TruncateBlocks to storage.Store with configurable retention window
- Wire cfg.Network.Namespace and cfg.Gateway.SyncTimeout through app.go
- Add run-demo.sh Ansible-based demo runner script
- Add USDC_deployment.md multi-replica deployment guide
- Extend replay_json_dataset_test.go with wrap-around support
- Add chain_test.go namespace-filter tests and store_test.go truncation tests

Signed-off-by: Eyal Kushnir <eyal.kushnir@ibm.com>

Author: kushnireyal

Makefile

@@ -24,6 +24,8 @@ build-release:
 			-o release/linux-$$arch/fxevm ./cmd/fxevm || exit 1; \
 	done
 
+IMAGE_TAG ?= dev
+
 .PHONY: build-image
 build-image: build-release
 	$(DOCKER) buildx build \
@@ -32,7 +34,7 @@ build-image: build-release
 		--build-arg VERSION=dev \
 		--build-arg CREATED=$(shell date -u +%Y-%m-%dT%H:%M:%SZ) \
 		--build-arg REVISION=$(shell git rev-parse HEAD) \
-		--tag fabric-x-evm:dev \
+		--tag fabric-x-evm:$(IMAGE_TAG) \
 		.
 
 .PHONY: checks
@@ -193,3 +195,16 @@ hardhat-tests:
 .PHONY: perf-tests
 perf-tests: pre-pull-images
 	@VERBOSE=$(VERBOSE) FABRIC_VERSION=$(FABRIC_VERSION) ./scripts/run_perf_test.sh
+
+# STAGING_HOST, SSH_USER, EVM_BRANCH, and DEMO_ARGS may be overridden on the command line.
+STAGING_HOST ?= dectrust8.vpc.cloud9.ibm.com
+SSH_USER     ?= root
+EVM_BRANCH   ?= $(shell git rev-parse --abbrev-ref HEAD)
+
+.PHONY: run-demo
+run-demo:
+	scripts/run-demo.sh \
+		--staging-host $(STAGING_HOST) \
+		--ssh-user $(SSH_USER) \
+		--evm-branch $(EVM_BRANCH) \
+		$(DEMO_ARGS)

gateway/app/app.go

@@ -114,7 +114,13 @@ func buildApp(ctx context.Context, cfg config.Config, gwSigner sdk.Signer, logge
 		return nil, fmt.Errorf("failed to create submitter: %w", err)
 	}
 
-	chain, err := core.NewChain(cfg.Gateway.Database.ConnString, cfg.Gateway.Database.TriePath, false)
+	chain, err := core.NewChain(core.ChainOpts{
+		DBConnString:            cfg.Gateway.Database.ConnString,
+		TriePath:                cfg.Gateway.Database.TriePath,
+		Namespace:               cfg.Network.Namespace,
+		BlockRetention:          cfg.Gateway.BlockRetention,
+		BlockTruncationInterval: cfg.Gateway.BlockTruncationInterval,
+	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to create chain: %w", err)
 	}
@@ -203,8 +209,12 @@ func (a *App) Run(ctx context.Context) error {
 	g.Go(func() error { return a.gwSync.Start(gctx) })
 
 	// Wait for initial sync before serving traffic
+	syncTimeout := a.cfg.Gateway.SyncTimeout
+	if syncTimeout == 0 {
+		syncTimeout = 10 * time.Second
+	}
 	for i, sync := range a.endorserSyncs {
-		if err := waitUntilSynced(gctx, sync, 10*time.Second); err != nil {
+		if err := waitUntilSynced(gctx, sync, syncTimeout); err != nil {
 			return err
 		}
 		appLogger.Debugf("endorser %d synced", i)

gateway/config/config.go

@@ -37,6 +37,9 @@ type Gateway struct {
 	EnableTestRPC    bool   `mapstructure:"enable-test-rpc"    yaml:"enable-test-rpc"`    // Enable test-only RPC methods (eth_accounts, eth_sendTransaction) - UNSAFE for production
 
 	WorkerCount int `mapstructure:"worker-count" yaml:"worker-count"` // number of worker goroutines; defaults to 1 if not set
+
+	BlockRetention          int    `mapstructure:"block-retention"           yaml:"block-retention"`           // number of recent blocks to keep on the gateway DB; 0 uses storage default
+	BlockTruncationInterval uint64 `mapstructure:"block-truncation-interval" yaml:"block-truncation-interval"` // run truncation every N blocks; 0 uses storage default
 }
 
 // DB holds the database paths for the gateway.

gateway/core/api.go

@@ -394,7 +394,7 @@ func (g *Gateway) Stop() error {
 
 	total, invalid := g.TxQueue.Stats()
 	if total > 0 {
-		fmt.Println("gw stats:", total, invalid, float64(invalid)/float64(total))
+		logger.Infof("gw stats: total=%d invalid=%d invalid_ratio=%.4f", total, invalid, float64(invalid)/float64(total))
 	}
 
 	return err

gateway/core/chain.go

@@ -28,22 +28,49 @@ import (
 // (for block ingestion) and core.Store (via the embedded *storage.Store, for API queries).
 type Chain struct {
 	*storage.Store
-	db       *sql.DB
-	ts       *trie.Store
-	prevHash common.Hash // Ethereum hash of last committed block; seeded from DB on startup
+	db        *sql.DB
+	ts        *trie.Store
+	prevHash  common.Hash // Ethereum hash of last committed block; seeded from DB on startup
+	namespace string      // Fabric namespace to filter; empty means accept all
+}
+
+// ChainOpts is the parameter bundle for NewChain.
+//
+// DBConnString uses the modernc SQLite DSN format.
+// TriePath is the directory for the PebbleDB trie; empty = in-memory (and ignored when WithTrie is false).
+// Namespace filters delivered blocks to only those containing transactions for that Fabric
+// namespace; empty accepts all namespaces.
+// BlockRetention and BlockTruncationInterval are forwarded to the underlying Store; zero values
+// use the storage package defaults (storage.DefaultBlockRetention / storage.DefaultBlockTruncationInterval).
+type ChainOpts struct {
+	DBConnString            string
+	TriePath                string
+	Namespace               string
+	WithTrie                bool
+	BlockRetention          int
+	BlockTruncationInterval uint64
 }
 
 // NewChain opens the SQLite database and trie store, seeds state from the latest committed
-// block, and returns a ready Chain. dbConnStr uses the modernc SQLite DSN format;
-// triePath is the directory for the PebbleDB trie (empty string = in-memory).
+// block, and returns a ready Chain.
 // The caller must register the SQLite driver (e.g. _ "modernc.org/sqlite") before calling.
-func NewChain(dbConnStr, triePath string, withTrie bool) (*Chain, error) {
-	db, err := sqlite.Open(dbConnStr)
+func NewChain(opts ChainOpts) (*Chain, error) {
+	db, err := sqlite.Open(opts.DBConnString)
 	if err != nil {
 		return nil, fmt.Errorf("open db: %w", err)
 	}
+	// Store SQLite temp objects in memory. The release image is built from
+	// scratch and has no /tmp, so file-based temp storage would fail with
+	// SQLITE_IOERR_GETTEMPPATH (6410) during WAL operations like DELETE.
+	if _, err := db.Exec("PRAGMA temp_store=MEMORY"); err != nil {
+		db.Close()
+		return nil, fmt.Errorf("set temp_store pragma: %w", err)
+	}
 
-	blockStore := storage.NewStore(db)
+	blockStore := storage.NewStore(db, storage.StoreOpts{
+		BlockRetention:          opts.BlockRetention,
+		BlockTruncationInterval: opts.BlockTruncationInterval,
+	})
 	if err := blockStore.Init(); err != nil {
 		db.Close()
 		return nil, fmt.Errorf("init block store: %w", err)
@@ -58,21 +85,39 @@ func NewChain(dbConnStr, triePath string, withTrie bool) (*Chain, error) {
 	}
 
 	var ts *trie.Store
-	if withTrie {
-		ts, err = trie.New(triePath, initialRoot)
+	if opts.WithTrie {
+		ts, err = trie.New(opts.TriePath, initialRoot)
 		if err != nil {
 			db.Close()
 			return nil, fmt.Errorf("open trie store: %w", err)
 		}
 	}
 
-	return &Chain{Store: blockStore, db: db, ts: ts, prevHash: prevHash}, nil
+	return &Chain{Store: blockStore, db: db, ts: ts, prevHash: prevHash, namespace: opts.Namespace}, nil
+}
+
+// filterAndConvert applies namespace filtering then delegates to the package-level ConvertToDomain.
+func (c *Chain) filterAndConvert(b blocks.Block) domain.Block {
+	if c.namespace == "" {
+		return ConvertToDomain(b)
+	}
+	filtered := b
+	filtered.Transactions = filtered.Transactions[:0:0]
+	for _, tx := range b.Transactions {
+		for _, nrws := range tx.NsRWS {
+			if nrws.Namespace == c.namespace {
+				filtered.Transactions = append(filtered.Transactions, tx)
+				break
+			}
+		}
+	}
+	return ConvertToDomain(filtered)
 }
 
 // Handle implements blocks.BlockHandler. It commits the block's write sets to the trie,
 // then persists the block and its transactions to the database.
 func (c *Chain) Handle(ctx context.Context, b blocks.Block) error {
-	ebl := ConvertToDomain(b)
+	ebl := c.filterAndConvert(b)
 
 	ebl.ParentHash = c.prevHash.Bytes()
 	if c.ts != nil {
@@ -90,6 +135,12 @@ func (c *Chain) Handle(ctx context.Context, b blocks.Block) error {
 		return err
 	}
 
+	if c.Store.BlockRetention > 0 && c.Store.BlockTruncationInterval > 0 && ebl.BlockNumber%c.Store.BlockTruncationInterval == 0 {
+		if err := c.Store.TruncateBlocks(ctx, c.Store.BlockRetention); err != nil {
+			logger.Warnf("block DB truncation failed at block %d: %v", ebl.BlockNumber, err)
+		}
+	}
+
 	return nil
 }
 
@@ -115,8 +166,6 @@ func ConvertToDomain(b blocks.Block) domain.Block {
 
 	logIndex := int64(0) // logIndex is the index of the log in the block
 	for _, tx := range b.Transactions {
-		// TODO: filter on namespace?
-
 		// retrieve the Ethereum transaction from the chaincode invocation
 		if len(tx.InputArgs) < 2 || !bytes.Equal(tx.InputArgs[0], []byte{byte(fc.ProposalTypeEVMTx)}) {
 			// skip non-eth tx

gateway/core/chain_test.go

@@ -129,6 +129,49 @@ func TestConvertToDomain_EmptyBlock(t *testing.T) {
 	assert.Len(t, got.Transactions, 0)
 }
 
+func TestConvertToDomain_NamespaceFilter(t *testing.T) {
+	key, err := crypto.GenerateKey()
+	require.NoError(t, err)
+	ethb := marshaledEthTx(t, key, common.HexToAddress("0x1111111111111111111111111111111111111111"), big.NewInt(1))
+
+	matchTx := blocks.Transaction{
+		ID:        "tx-match",
+		Valid:     true,
+		InputArgs: [][]byte{{byte(co.ProposalTypeEVMTx)}, ethb},
+		NsRWS:     []blocks.NsReadWriteSet{{Namespace: "my-ns"}},
+	}
+	foreignTx := blocks.Transaction{
+		ID:        "tx-foreign",
+		Valid:     true,
+		InputArgs: [][]byte{{byte(co.ProposalTypeEVMTx)}, ethb},
+		NsRWS:     []blocks.NsReadWriteSet{{Namespace: "other-ns"}},
+	}
+
+	t.Run("filters out foreign namespace", func(t *testing.T) {
+		c := &Chain{namespace: "my-ns"}
+		got := c.filterAndConvert(blocks.Block{Number: 1, Transactions: []blocks.Transaction{matchTx, foreignTx}})
+		require.Len(t, got.Transactions, 1)
+		assert.Equal(t, "tx-match", got.Transactions[0].FabricTxID)
+	})
+
+	t.Run("no-op when namespace is empty", func(t *testing.T) {
+		c := &Chain{}
+		got := c.filterAndConvert(blocks.Block{Number: 1, Transactions: []blocks.Transaction{matchTx, foreignTx}})
+		assert.Len(t, got.Transactions, 2)
+	})
+
+	t.Run("skips tx with no NsRWS when namespace set", func(t *testing.T) {
+		noNsTx := blocks.Transaction{
+			ID:        "tx-no-ns",
+			Valid:     true,
+			InputArgs: [][]byte{{byte(co.ProposalTypeEVMTx)}, ethb},
+		}
+		c := &Chain{namespace: "my-ns"}
+		got := c.filterAndConvert(blocks.Block{Number: 1, Transactions: []blocks.Transaction{noNsTx}})
+		assert.Len(t, got.Transactions, 0)
+	})
+}
+
 // --- convertTransaction ---
 
 func TestConvertTransaction_RegularTransfer(t *testing.T) {

gateway/core/txqueue_helpers.go

@@ -43,9 +43,13 @@ func senderForTx(tx *types.Transaction) (common.Address, bool) {
 	return sender, true
 }
 
-// recipientForTx extracts the recipient address from a transaction.
-// For ERC20 transfers, it extracts the recipient from the calldata.
-// Returns (address, true) if successful, (zero address, false) otherwise.
+// erc20TransferSelector is the 4-byte function selector for ERC20 transfer(address,uint256).
+var erc20TransferSelector = [4]byte{0xa9, 0x05, 0x9c, 0xbb}
+
+// recipientForTx extracts the recipient address from an ERC20 transfer(address,uint256) call.
+// Returns (address, true) only when the calldata matches the ERC20 transfer selector and is the
+// expected length. Without the selector check, any call with >=68 bytes of data would be decoded
+// as a transfer and create false-positive dependency edges between unrelated transactions.
 func recipientForTx(tx *types.Transaction) (common.Address, bool) {
 	if tx.To() == nil {
 		return common.Address{}, false
@@ -55,6 +59,10 @@ func recipientForTx(tx *types.Transaction) (common.Address, bool) {
 	if len(data) < 4+32+32 {
 		return common.Address{}, false
 	}
+	if data[0] != erc20TransferSelector[0] || data[1] != erc20TransferSelector[1] ||
+		data[2] != erc20TransferSelector[2] || data[3] != erc20TransferSelector[3] {
+		return common.Address{}, false
+	}
 
 	// Extract recipient from ERC20 transfer calldata (offset 4 + 12 bytes)
 	recipientOffset := 4 + 12

gateway/storage/store.go

@@ -21,15 +21,37 @@ import (
 var ddl string
 
 type Store struct {
-	queries           *Queries
-	DB                *sql.DB
-	CachedBlockNumber atomic.Uint64 // cached block number for fast reads
+	queries                 *Queries
+	DB                      *sql.DB
+	CachedBlockNumber       atomic.Uint64 // cached block number for fast reads
+	BlockRetention          int           // number of recent blocks to keep; 0 disables truncation
+	BlockTruncationInterval uint64        // run truncation every N blocks; 0 disables truncation
 }
 
-func NewStore(db *sql.DB) *Store {
+// StoreOpts carries optional tuning for NewStore. Zero values use built-in defaults
+// (DefaultBlockRetention / DefaultBlockTruncationInterval).
+type StoreOpts struct {
+	BlockRetention          int
+	BlockTruncationInterval uint64
+}
+
+const (
+	DefaultBlockRetention          = 10000
+	DefaultBlockTruncationInterval = 1000
+)
+
+func NewStore(db *sql.DB, opts StoreOpts) *Store {
+	if opts.BlockRetention == 0 {
+		opts.BlockRetention = DefaultBlockRetention
+	}
+	if opts.BlockTruncationInterval == 0 {
+		opts.BlockTruncationInterval = DefaultBlockTruncationInterval
+	}
 	return &Store{
-		queries: New(db),
-		DB:      db,
+		queries:                 New(db),
+		DB:                      db,
+		BlockRetention:          opts.BlockRetention,
+		BlockTruncationInterval: opts.BlockTruncationInterval,
 	}
 }
 
@@ -451,6 +473,34 @@ func (s *Store) GetLogs(ctx context.Context, filter domain.LogFilter) ([]domain.
 	return logs, nil
 }
 
+// TruncateBlocks deletes blocks older than the last keepLastN blocks, along with
+// their associated transactions and logs. All three tables are deleted in a single
+// transaction to maintain referential integrity (logs and transactions have foreign
+// keys on block_number with no CASCADE, so they must be deleted first).
+// Returns nil immediately if there is nothing to truncate.
+func (s *Store) TruncateBlocks(ctx context.Context, keepLastN int) error {
+	cutoff := int64(s.CachedBlockNumber.Load()) - int64(keepLastN)
+	if cutoff <= 0 {
+		return nil
+	}
+	sqlTx, err := s.DB.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer sqlTx.Rollback() //nolint:errcheck
+
+	for _, stmt := range []string{
+		`DELETE FROM logs WHERE block_number <= ?`,
+		`DELETE FROM transactions WHERE block_number <= ?`,
+		`DELETE FROM blocks WHERE block_number <= ?`,
+	} {
+		if _, err := sqlTx.Exec(stmt, cutoff); err != nil {
+			return err
+		}
+	}
+	return sqlTx.Commit()
+}
+
 // GetLogsByTxHash retrieves all logs for a specific transaction.
 func (s *Store) GetLogsByTxHash(ctx context.Context, txHash []byte) ([]domain.Log, error) {
 	rows, err := s.queries.GetLogsByTxHash(ctx, txHash)