Add namespace-filtered delivery and block DB retention for multi-replica demo

Each EVM gateway replica now skips transactions from foreign Fabric namespaces
(prevents O(N) delivery cost in a 5-replica setup). Adds TruncateBlocks to
prune SQLite to a configurable retention window, triggered every 1000 blocks.
Also adds USDC demo deployment runbook in integration/perf/USDC_deployment.md.

Signed-off-by: Eyal Kushnir <eyalk7@gmail.com>
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,7 @@ func buildApp(cfg config.Config, gwSigner sdk.Signer, logger sdk.Logger, endorse
 		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(cfg.Gateway.Database.ConnString, cfg.Gateway.Database.TriePath, cfg.Network.Namespace, false)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create chain: %w", err)
 	}
@@ -196,8 +196,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 = 5 * time.Minute
+	}
 	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/core/api.go

@@ -100,6 +100,15 @@ func (g *Gateway) Start(ctx context.Context) {
 // worker processes transactions from the queue
 func (g *Gateway) worker(ctx context.Context) {
 	defer g.wg.Done()
+	defer func() {
+		if r := recover(); r != nil {
+			logger.Errorf("worker panic: %v", r)
+			if ctx.Err() == nil {
+				g.wg.Add(1)
+				go g.worker(ctx)
+			}
+		}
+	}()
 
 	for {
 		tx, ok := g.TxQueue.Dequeue()

gateway/core/chain.go

@@ -28,20 +28,30 @@ 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
 }
 
 // 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).
+// namespace filters delivered blocks to only those containing transactions for that Fabric
+// namespace; pass an empty string to accept all namespaces.
 // The caller must register the SQLite driver (e.g. _ "modernc.org/sqlite") before calling.
-func NewChain(dbConnStr, triePath string, withTrie bool) (*Chain, error) {
+func NewChain(dbConnStr, triePath, namespace string, withTrie bool) (*Chain, error) {
 	db, err := sqlite.Open(dbConnStr)
 	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)
 	if err := blockStore.Init(); err != nil {
@@ -66,13 +76,31 @@ func NewChain(dbConnStr, triePath string, withTrie bool) (*Chain, error) {
 		}
 	}
 
-	return &Chain{Store: blockStore, db: db, ts: ts, prevHash: prevHash}, nil
+	return &Chain{Store: blockStore, db: db, ts: ts, prevHash: prevHash, namespace: namespace}, nil
+}
+
+// convertToDomain applies namespace filtering then delegates to ConvertToDomain.
+func (c *Chain) convertToDomain(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.convertToDomain(b)
 
 	ebl.ParentHash = c.prevHash.Bytes()
 	if c.ts != nil {
@@ -90,6 +118,12 @@ func (c *Chain) Handle(ctx context.Context, b blocks.Block) error {
 		return err
 	}
 
+	if c.Store.BlockRetention > 0 && ebl.BlockNumber%1000 == 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 +149,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
@@ -129,7 +161,8 @@ func ConvertToDomain(b blocks.Block) domain.Block {
 
 		etx, err := convertTransaction(tx.InputArgs[1], b.Hash, b.Number, tx.Number, tx.ID, status, tx.Status, tx.Events, &logIndex)
 		if err != nil {
-			panic(err) // we surface this for now instead of swallowing it
+			logger.Warnf("skipping malformed tx %s in block %d: %v", tx.ID, b.Number, err)
+			continue
 		}
 
 		ebl.Transactions = append(ebl.Transactions, etx)

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.convertToDomain(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.convertToDomain(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.convertToDomain(blocks.Block{Number: 1, Transactions: []blocks.Transaction{noNsTx}})
+		assert.Len(t, got.Transactions, 0)
+	})
+}
+
 // --- convertTransaction ---
 
 func TestConvertTransaction_RegularTransfer(t *testing.T) {

gateway/storage/store.go

@@ -24,12 +24,16 @@ type Store struct {
 	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
 }
 
+const DefaultBlockRetention = 10000
+
 func NewStore(db *sql.DB) *Store {
 	return &Store{
-		queries: New(db),
-		DB:      db,
+		queries:        New(db),
+		DB:             db,
+		BlockRetention: DefaultBlockRetention,
 	}
 }
 
@@ -451,6 +455,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)

gateway/storage/store_test.go

@@ -764,6 +764,79 @@ func TestInsertBlock_WithTransactionsAndLogs(t *testing.T) {
 	}
 }
 
+// TruncateBlocks tests
+
+func TestTruncateBlocks(t *testing.T) {
+	store := setupTestDB(t)
+
+	// Insert blocks 1-10, each with a transaction and a log
+	for i := uint64(1); i <= 10; i++ {
+		bh := makeHash(byte(i))
+		insertTestBlock(t, store, i, bh)
+		th := makeHash(byte(i + 100))
+		insertTestTransaction(t, store, i, bh, th, 0)
+		insertTestLog(t, store, i, th, makeAddress(0x01), 0, nil)
+	}
+
+	// Keep last 5: blocks 6-10 should survive, 1-5 should be deleted
+	err := store.TruncateBlocks(t.Context(), 5)
+	if err != nil {
+		t.Fatalf("TruncateBlocks error: %v", err)
+	}
+
+	// Block 5 should be gone
+	b, err := store.GetBlockByNumber(t.Context(), 5, false)
+	if err != nil {
+		t.Fatalf("GetBlockByNumber error: %v", err)
+	}
+	if b != nil {
+		t.Error("expected block 5 to be deleted")
+	}
+
+	// Block 6 should still exist
+	b, err = store.GetBlockByNumber(t.Context(), 6, false)
+	if err != nil {
+		t.Fatalf("GetBlockByNumber error: %v", err)
+	}
+	if b == nil {
+		t.Error("expected block 6 to survive")
+	}
+
+	// Transactions for block 5 should be gone
+	tx, err := store.GetTransactionByBlockNumberAndIndex(t.Context(), 5, 0)
+	if err != nil {
+		t.Fatalf("GetTransactionByBlockNumberAndIndex error: %v", err)
+	}
+	if tx != nil {
+		t.Error("expected transaction for block 5 to be deleted")
+	}
+
+	// Logs for blocks 1-5 should be gone
+	five := uint64(5)
+	logs, err := store.GetLogs(t.Context(), domain.LogFilter{ToBlock: &five})
+	if err != nil {
+		t.Fatalf("GetLogs error: %v", err)
+	}
+	if len(logs) != 0 {
+		t.Errorf("expected 0 logs for blocks 1-5, got %d", len(logs))
+	}
+}
+
+func TestTruncateBlocks_NothingToDelete(t *testing.T) {
+	store := setupTestDB(t)
+	for i := uint64(1); i <= 3; i++ {
+		insertTestBlock(t, store, i, makeHash(byte(i)))
+	}
+	// keepLastN > total blocks: nothing should be deleted
+	if err := store.TruncateBlocks(t.Context(), 100); err != nil {
+		t.Fatalf("TruncateBlocks error: %v", err)
+	}
+	b, _ := store.GetBlockByNumber(t.Context(), 1, false)
+	if b == nil {
+		t.Error("block 1 should not have been deleted")
+	}
+}
+
 // GetLogsByTxHash test
 
 func TestGetLogsByTxHash(t *testing.T) {