handle updated accounts in shared map

there's a possibly (though yet unseen) that an updated account could be moved outside the domain of
a bounded iterater, in which case it would not see the update after traversal. previously this would
have caused an error, but this should prevent it from happening.

builder.go

@@ -69,8 +69,20 @@ type accountUpdate struct {
 	new     sdtypes.AccountWrapper
 	oldRoot common.Hash
 }
+
 type accountUpdateMap map[string]*accountUpdate
 
+type accountUpdateLens struct {
+	state accountUpdateMap
+	sync.Mutex
+}
+
+func (l *accountUpdateLens) update(fn func(accountUpdateMap)) {
+	l.Lock()
+	defer l.Unlock()
+	fn(l.state)
+}
+
 func appender[T any](to *[]T) func(T) error {
 	return func(a T) error {
 		*to = append(*to, a)
@@ -141,8 +153,11 @@ func (sdb *builder) WriteStateDiff(
 	subitersA := iterutils.SubtrieIterators(triea.NodeIterator, uint(sdb.subtrieWorkers))
 	subitersB := iterutils.SubtrieIterators(trieb.NodeIterator, uint(sdb.subtrieWorkers))
 
+	updates := accountUpdateLens{
+		state: make(accountUpdateMap),
+	}
 	logger := log.New("hash", args.BlockHash, "number", args.BlockNumber)
-	// errgroup will cancel if any group fails
+	// errgroup will cancel if any worker fails
 	g, ctx := errgroup.WithContext(context.Background())
 	for i := uint(0); i < sdb.subtrieWorkers; i++ {
 		func(subdiv uint) {
@@ -152,12 +167,35 @@ func (sdb *builder) WriteStateDiff(
 				return sdb.processAccounts(ctx,
 					it, &it.SymmDiffState,
 					params.watchedAddressesLeafPaths,
-					nodeSink, ipldSink, logger,
+					nodeSink, ipldSink, &updates,
+					logger,
 				)
 			})
 		}(i)
 	}
-	return g.Wait()
+
+	if err = g.Wait(); err != nil {
+		return err
+	}
+
+	for key, update := range updates.state {
+		var storageDiff []sdtypes.StorageLeafNode
+		err := sdb.processStorageUpdates(
+			update.oldRoot, update.new.Account.Root,
+			appender(&storageDiff), ipldSink,
+		)
+		if err != nil {
+			return fmt.Errorf("error processing incremental storage diffs for account with leafkey %x\r\nerror: %w", key, err)
+		}
+
+		if err = nodeSink(sdtypes.StateLeafNode{
+			AccountWrapper: update.new,
+			StorageDiff:    storageDiff,
+		}); err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
 // WriteStateDiff writes a statediff object to output sinks
@@ -191,7 +229,11 @@ func (sdb *builder) WriteStateSnapshot(
 			subiters[i] = tracker.Tracked(subiters[i])
 		}
 	}
-	// errgroup will cancel if any group fails
+	updates := accountUpdateLens{
+		state: make(accountUpdateMap),
+	}
+
+	// errgroup will cancel if any worker fails
 	g, ctx := errgroup.WithContext(context.Background())
 	for i := range subiters {
 		func(subdiv uint) {
@@ -200,7 +242,8 @@ func (sdb *builder) WriteStateSnapshot(
 				return sdb.processAccounts(ctx,
 					subiters[subdiv], &symdiff,
 					params.watchedAddressesLeafPaths,
-					nodeSink, ipldSink, log.DefaultLogger,
+					nodeSink, ipldSink, &updates,
+					log.DefaultLogger,
 				)
 			})
 		}(uint(i))
@@ -215,13 +258,13 @@ func (sdb *builder) processAccounts(
 	it trie.NodeIterator, symdiff *utils.SymmDiffState,
 	watchedAddressesLeafPaths [][]byte,
 	nodeSink sdtypes.StateNodeSink, ipldSink sdtypes.IPLDSink,
+	updateLens *accountUpdateLens,
 	logger log.Logger,
 ) error {
 	logger.Trace("statediff/processAccounts BEGIN")
 	defer metrics.ReportAndUpdateDuration("statediff/processAccounts END",
 		time.Now(), logger, metrics.IndexerMetrics.ProcessAccountsTimer)
 
-	updates := make(accountUpdateMap)
 	// Cache the RLP of the previous node. When we hit a value node this will be the parent blob.
 	var prevBlob = it.NodeBlob()
 	for it.Next(true) {
@@ -245,12 +288,14 @@ func (sdb *builder) processAccounts(
 				copy(leafKey, it.LeafKey())
 
 				if symdiff.CommonPath() {
-					// If B also contains this leaf node, this is the old state of an updated account.
-					if update, ok := updates[string(leafKey)]; ok {
-						update.oldRoot = account.Root
-					} else {
-						updates[string(leafKey)] = &accountUpdate{oldRoot: account.Root}
-					}
+					updateLens.update(func(updates accountUpdateMap) {
+						// If B also contains this leaf node, this is the old state of an updated account.
+						if update, ok := updates[string(leafKey)]; ok {
+							update.oldRoot = account.Root
+						} else {
+							updates[string(leafKey)] = &accountUpdate{oldRoot: account.Root}
+						}
+					})
 				} else {
 					// This node was removed, meaning the account was deleted.  Emit empty
 					// "removed" records for the state node and all storage all storage slots.
@@ -270,12 +315,14 @@ func (sdb *builder) processAccounts(
 			}
 
 			if symdiff.CommonPath() {
-				// If A also contains this leaf node, this is the new state of an updated account.
-				if update, ok := updates[string(accountW.LeafKey)]; ok {
-					update.new = *accountW
-				} else {
-					updates[string(accountW.LeafKey)] = &accountUpdate{new: *accountW}
-				}
+				updateLens.update(func(updates accountUpdateMap) {
+					// If A also contains this leaf node, this is the new state of an updated account.
+					if update, ok := updates[string(accountW.LeafKey)]; ok {
+						update.new = *accountW
+					} else {
+						updates[string(accountW.LeafKey)] = &accountUpdate{new: *accountW}
+					}
+				})
 			} else { // account was created
 				err := sdb.processAccountCreation(accountW, ipldSink, nodeSink)
 				if err != nil {
@@ -318,24 +365,6 @@ func (sdb *builder) processAccounts(
 		}
 		prevBlob = nodeVal
 	}
-
-	for key, update := range updates {
-		var storageDiff []sdtypes.StorageLeafNode
-		err := sdb.processStorageUpdates(
-			update.oldRoot, update.new.Account.Root,
-			appender(&storageDiff), ipldSink,
-		)
-		if err != nil {
-			return fmt.Errorf("error processing incremental storage diffs for account with leafkey %x\r\nerror: %w", key, err)
-		}
-
-		if err = nodeSink(sdtypes.StateLeafNode{
-			AccountWrapper: update.new,
-			StorageDiff:    storageDiff,
-		}); err != nil {
-			return err
-		}
-	}
 	return it.Error()
 }
 

indexer/constructor.go

@@ -33,11 +33,19 @@ import (
 )
 
 // NewStateDiffIndexer creates and returns an implementation of the StateDiffIndexer interface.
+// The returned indexer is used to process state diffs and write them to a database.
+// The returned SQL database, if non-nil, is the indexer's backing database.
+// `ctx` is used to cancel an underlying DB connection.
+// `chainConfig` is used when processing chain state.
+// `nodeInfo` contains metadata on the Ethereum node, which is inserted with the indexed state.
+// `config` contains configuration specific to the indexer.
+// `isDiff` means to mark state nodes as belonging to an incremental diff, as opposed to a full snapshot.
 func NewStateDiffIndexer(
 	ctx context.Context,
 	chainConfig *params.ChainConfig,
 	nodeInfo node.Info,
 	config interfaces.Config,
+	isDiff bool,
 ) (
 	sql.Database,
 	interfaces.StateDiffIndexer,
@@ -50,7 +58,7 @@ func NewStateDiffIndexer(
 		if !ok {
 			return nil, nil, fmt.Errorf("file config is not the correct type: got %T, expected %T", config, file.Config{})
 		}
-		ind, err := file.NewStateDiffIndexer(chainConfig, fc, nodeInfo)
+		ind, err := file.NewStateDiffIndexer(chainConfig, fc, nodeInfo, isDiff)
 		return nil, ind, err
 	case shared.POSTGRES:
 		log.Info("Starting statediff service in Postgres writing mode")
@@ -75,7 +83,7 @@ func NewStateDiffIndexer(
 			return nil, nil, fmt.Errorf("unrecognized Postgres driver type: %s", pgc.Driver)
 		}
 		db := postgres.NewPostgresDB(driver, pgc.Upsert)
-		ind, err := sql.NewStateDiffIndexer(ctx, chainConfig, db)
+		ind, err := sql.NewStateDiffIndexer(ctx, chainConfig, db, isDiff)
 		return db, ind, err
 	case shared.DUMP:
 		log.Info("Starting statediff service in data dump mode")

indexer/database/file/csv_indexer_legacy_test.go

@@ -43,7 +43,7 @@ func setupLegacyCSVIndexer(t *testing.T) {
 		require.NoError(t, err)
 	}
 
-	ind, err = file.NewStateDiffIndexer(test.LegacyConfig, file.CSVTestConfig, test.LegacyNodeInfo)
+	ind, err = file.NewStateDiffIndexer(test.LegacyConfig, file.CSVTestConfig, test.LegacyNodeInfo, true)
 	require.NoError(t, err)
 
 	db, err = postgres.SetupSQLXDB()

indexer/database/file/csv_indexer_test.go

@@ -41,7 +41,7 @@ func setupCSVIndexer(t *testing.T) {
 		require.NoError(t, err)
 	}
 
-	ind, err = file.NewStateDiffIndexer(mocks.TestChainConfig, file.CSVTestConfig, test.LegacyNodeInfo)
+	ind, err = file.NewStateDiffIndexer(mocks.TestChainConfig, file.CSVTestConfig, test.LegacyNodeInfo, true)
 	require.NoError(t, err)
 
 	db, err = postgres.SetupSQLXDB()

indexer/database/file/csv_writer.go

@@ -57,7 +57,8 @@ type tableRow struct {
 
 type CSVWriter struct {
 	// dir containing output files
-	dir string
+	dir    string
+	isDiff bool
 
 	writers                fileWriters
 	watchedAddressesWriter fileWriter
@@ -128,7 +129,7 @@ func (tx fileWriters) flush() error {
 	return nil
 }
 
-func NewCSVWriter(path string, watchedAddressesFilePath string) (*CSVWriter, error) {
+func NewCSVWriter(path string, watchedAddressesFilePath string, diff bool) (*CSVWriter, error) {
 	if err := os.MkdirAll(path, 0777); err != nil {
 		return nil, fmt.Errorf("unable to create directory '%s': %w", path, err)
 	}
@@ -147,6 +148,7 @@ func NewCSVWriter(path string, watchedAddressesFilePath string) (*CSVWriter, err
 		writers:                writers,
 		watchedAddressesWriter: watchedAddressesWriter,
 		dir:                    path,
+		isDiff:                 diff,
 		rows:                   make(chan tableRow),
 		flushChan:              make(chan struct{}),
 		flushFinished:          make(chan struct{}),
@@ -278,14 +280,14 @@ func (csw *CSVWriter) upsertStateCID(stateNode models.StateNodeModel) {
 
 	var values []interface{}
 	values = append(values, stateNode.BlockNumber, stateNode.HeaderID, stateNode.StateKey, stateNode.CID,
-		true, balance, strconv.FormatUint(stateNode.Nonce, 10), stateNode.CodeHash, stateNode.StorageRoot, stateNode.Removed)
+		csw.isDiff, balance, strconv.FormatUint(stateNode.Nonce, 10), stateNode.CodeHash, stateNode.StorageRoot, stateNode.Removed)
 	csw.rows <- tableRow{schema.TableStateNode, values}
 }
 
 func (csw *CSVWriter) upsertStorageCID(storageCID models.StorageNodeModel) {
 	var values []interface{}
 	values = append(values, storageCID.BlockNumber, storageCID.HeaderID, storageCID.StateKey, storageCID.StorageKey, storageCID.CID,
-		true, storageCID.Value, storageCID.Removed)
+		csw.isDiff, storageCID.Value, storageCID.Removed)
 	csw.rows <- tableRow{schema.TableStorageNode, values}
 }
 

indexer/database/file/indexer.go

@@ -63,7 +63,11 @@ type StateDiffIndexer struct {
 }
 
 // NewStateDiffIndexer creates a void implementation of interfaces.StateDiffIndexer
-func NewStateDiffIndexer(chainConfig *params.ChainConfig, config Config, nodeInfo node.Info) (*StateDiffIndexer, error) {
+// `chainConfig` is used when processing chain state.
+// `config` contains configuration specific to the indexer.
+// `nodeInfo` contains metadata on the Ethereum node, which is inserted with the indexed state.
+// `isDiff` means to mark state nodes as belonging to an incremental diff, as opposed to a full snapshot.
+func NewStateDiffIndexer(chainConfig *params.ChainConfig, config Config, nodeInfo node.Info, isDiff bool) (*StateDiffIndexer, error) {
 	var err error
 	var writer FileWriter
 
@@ -86,7 +90,7 @@ func NewStateDiffIndexer(chainConfig *params.ChainConfig, config Config, nodeInf
 		}
 		log.Info("Writing watched addresses to file", "file", watchedAddressesFilePath)
 
-		writer, err = NewCSVWriter(outputDir, watchedAddressesFilePath)
+		writer, err = NewCSVWriter(outputDir, watchedAddressesFilePath, isDiff)
 		if err != nil {
 			return nil, err
 		}
@@ -109,7 +113,7 @@ func NewStateDiffIndexer(chainConfig *params.ChainConfig, config Config, nodeInf
 		}
 		log.Info("Writing watched addresses to file", "file", watchedAddressesFilePath)
 
-		writer = NewSQLWriter(file, watchedAddressesFilePath)
+		writer = NewSQLWriter(file, watchedAddressesFilePath, isDiff)
 	default:
 		return nil, fmt.Errorf("unrecognized file mode: %s", config.Mode)
 	}

indexer/database/file/mainnet_tests/indexer_test.go

@@ -83,7 +83,7 @@ func setupMainnetIndexer(t *testing.T) {
 		require.NoError(t, err)
 	}
 
-	ind, err = file.NewStateDiffIndexer(chainConf, file.CSVTestConfig, test.LegacyNodeInfo)
+	ind, err = file.NewStateDiffIndexer(chainConf, file.CSVTestConfig, test.LegacyNodeInfo, true)
 	require.NoError(t, err)
 
 	db, err = postgres.SetupSQLXDB()

indexer/database/file/sql_indexer_legacy_test.go

@@ -44,7 +44,7 @@ func setupLegacySQLIndexer(t *testing.T) {
 		require.NoError(t, err)
 	}
 
-	ind, err = file.NewStateDiffIndexer(test.LegacyConfig, file.SQLTestConfig, test.LegacyNodeInfo)
+	ind, err = file.NewStateDiffIndexer(test.LegacyConfig, file.SQLTestConfig, test.LegacyNodeInfo, true)
 	require.NoError(t, err)
 
 	db, err = postgres.SetupSQLXDB()

indexer/database/file/sql_indexer_test.go

@@ -41,7 +41,7 @@ func setupIndexer(t *testing.T) {
 		require.NoError(t, err)
 	}
 
-	ind, err = file.NewStateDiffIndexer(mocks.TestChainConfig, file.SQLTestConfig, test.LegacyNodeInfo)
+	ind, err = file.NewStateDiffIndexer(mocks.TestChainConfig, file.SQLTestConfig, test.LegacyNodeInfo, true)
 	require.NoError(t, err)
 
 	db, err = postgres.SetupSQLXDB()

indexer/database/file/sql_writer.go

@@ -44,6 +44,7 @@ var (
 type SQLWriter struct {
 	wc             io.WriteCloser
 	stmts          chan []byte
+	isDiff         bool
 	collatedStmt   []byte
 	collationIndex int
 
@@ -55,11 +56,15 @@ type SQLWriter struct {
 	watchedAddressesFilePath string
 }
 
-// NewSQLWriter creates a new pointer to a Writer
-func NewSQLWriter(wc io.WriteCloser, watchedAddressesFilePath string) *SQLWriter {
+// NewSQLWriter creates a new Writer.
+// `wc` is the underlying io.WriteCloser to write to.
+// `watchedAddressesFilePath` is the path to the file containing watched addresses.
+// `isDiff` means to mark state nodes as belonging to an incremental diff, as opposed to a full snapshot.
+func NewSQLWriter(wc io.WriteCloser, watchedAddressesFilePath string, isDiff bool) *SQLWriter {
 	return &SQLWriter{
 		wc:                       wc,
 		stmts:                    make(chan []byte),
+		isDiff:                   isDiff,
 		collatedStmt:             make([]byte, writeBufferSize),
 		flushChan:                make(chan struct{}),
 		flushFinished:            make(chan struct{}),
@@ -225,12 +230,12 @@ func (sqw *SQLWriter) upsertStateCID(stateNode models.StateNodeModel) {
 		balance = "0"
 	}
 	sqw.stmts <- []byte(fmt.Sprintf(stateInsert, stateNode.BlockNumber, stateNode.HeaderID, stateNode.StateKey, stateNode.CID,
-		stateNode.Removed, true, balance, stateNode.Nonce, stateNode.CodeHash, stateNode.StorageRoot))
+		stateNode.Removed, sqw.isDiff, balance, stateNode.Nonce, stateNode.CodeHash, stateNode.StorageRoot))
 }
 
 func (sqw *SQLWriter) upsertStorageCID(storageCID models.StorageNodeModel) {
 	sqw.stmts <- []byte(fmt.Sprintf(storageInsert, storageCID.BlockNumber, storageCID.HeaderID, storageCID.StateKey, storageCID.StorageKey, storageCID.CID,
-		storageCID.Removed, true, storageCID.Value))
+		storageCID.Removed, sqw.isDiff, storageCID.Value))
 }
 
 // LoadWatchedAddresses loads watched addresses from a file

indexer/database/sql/indexer.go

@@ -43,7 +43,7 @@ import (
 
 var _ interfaces.StateDiffIndexer = &StateDiffIndexer{}
 
-// StateDiffIndexer satisfies the indexer.StateDiffIndexer interface for ethereum statediff objects on top of an SQL sql
+// StateDiffIndexer satisfies the indexer.StateDiffIndexer interface for ethereum statediff objects on top of an SQL DB.
 type StateDiffIndexer struct {
 	ctx         context.Context
 	chainConfig *params.ChainConfig
@@ -51,11 +51,17 @@ type StateDiffIndexer struct {
 }
 
 // NewStateDiffIndexer creates a sql implementation of interfaces.StateDiffIndexer
-func NewStateDiffIndexer(ctx context.Context, chainConfig *params.ChainConfig, db Database) (*StateDiffIndexer, error) {
+// `ctx` is used to cancel the underlying DB connection.
+// `chainConfig` is used when processing chain state.
+// `db` is the backing database to use.
+// `isDiff` means to mark state nodes as belonging to an incremental diff, as opposed to a full snapshot.
+func NewStateDiffIndexer(
+	ctx context.Context, chainConfig *params.ChainConfig, db Database, isDiff bool,
+) (*StateDiffIndexer, error) {
 	return &StateDiffIndexer{
 		ctx:         ctx,
 		chainConfig: chainConfig,
-		dbWriter:    NewWriter(db),
+		dbWriter:    NewWriter(db, isDiff),
 	}, nil
 }
 

indexer/database/sql/lazy_tx.go

@@ -3,6 +3,7 @@ package sql
 import (
 	"context"
 	"reflect"
+	"sync"
 	"time"
 
 	"github.com/cerc-io/plugeth-statediff/indexer/database/metrics"
@@ -15,6 +16,7 @@ const copyFromCheckLimit = 100
 type DelayedTx struct {
 	cache []interface{}
 	db    Database
+	sync.RWMutex
 }
 type cachedStmt struct {
 	sql  string
@@ -27,6 +29,8 @@ type copyFrom struct {
 	rows        [][]interface{}
 }
 
+type result int64
+
 func (cf *copyFrom) appendRows(rows [][]interface{}) {
 	cf.rows = append(cf.rows, rows...)
 }
@@ -44,6 +48,8 @@ func (tx *DelayedTx) QueryRow(ctx context.Context, sql string, args ...interface
 }
 
 func (tx *DelayedTx) findPrevCopyFrom(tableName []string, columnNames []string, limit int) (*copyFrom, int) {
+	tx.RLock()
+	defer tx.RUnlock()
 	for pos, count := len(tx.cache)-1, 0; pos >= 0 && count < limit; pos, count = pos-1, count+1 {
 		prevCopy, ok := tx.cache[pos].(*copyFrom)
 		if ok && prevCopy.matches(tableName, columnNames) {
@@ -59,15 +65,19 @@ func (tx *DelayedTx) CopyFrom(ctx context.Context, tableName []string, columnNam
 			"current", len(prevCopy.rows), "new", len(rows), "distance", distance)
 		prevCopy.appendRows(rows)
 	} else {
+		tx.Lock()
 		tx.cache = append(tx.cache, &copyFrom{tableName, columnNames, rows})
+		tx.Unlock()
 	}
 
 	return 0, nil
 }
 
 func (tx *DelayedTx) Exec(ctx context.Context, sql string, args ...interface{}) (Result, error) {
+	tx.Lock()
 	tx.cache = append(tx.cache, cachedStmt{sql, args})
-	return nil, nil
+	defer tx.Unlock()
+	return result(0), nil
 }
 
 func (tx *DelayedTx) Commit(ctx context.Context) error {
@@ -85,6 +95,8 @@ func (tx *DelayedTx) Commit(ctx context.Context) error {
 			rollback(ctx, base)
 		}
 	}()
+	tx.Lock()
+	defer tx.Unlock()
 	for _, item := range tx.cache {
 		switch item := item.(type) {
 		case *copyFrom:
@@ -105,6 +117,13 @@ func (tx *DelayedTx) Commit(ctx context.Context) error {
 }
 
 func (tx *DelayedTx) Rollback(ctx context.Context) error {
+	tx.Lock()
+	defer tx.Unlock()
 	tx.cache = nil
 	return nil
 }
+
+// RowsAffected satisfies sql.Result
+func (r result) RowsAffected() (int64, error) {
+	return int64(r), nil
+}

indexer/database/sql/mainnet_tests/indexer_test.go

@@ -71,7 +71,7 @@ func setupMainnetIndexer(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	ind, err = sql.NewStateDiffIndexer(context.Background(), chainConf, db)
+	ind, err = sql.NewStateDiffIndexer(context.Background(), chainConf, db, true)
 }
 
 func checkTxClosure(t *testing.T, idle, inUse, open int64) {

indexer/database/sql/pgx_indexer_legacy_test.go

@@ -33,7 +33,7 @@ func setupLegacyPGXIndexer(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	ind, err = sql.NewStateDiffIndexer(context.Background(), test.LegacyConfig, db)
+	ind, err = sql.NewStateDiffIndexer(context.Background(), test.LegacyConfig, db, true)
 	require.NoError(t, err)
 }
 

indexer/database/sql/pgx_indexer_test.go

@@ -44,7 +44,7 @@ func setupPGXIndexer(t *testing.T, config postgres.Config) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	ind, err = sql.NewStateDiffIndexer(context.Background(), mocks.TestChainConfig, db)
+	ind, err = sql.NewStateDiffIndexer(context.Background(), mocks.TestChainConfig, db, true)
 	require.NoError(t, err)
 }
 

indexer/database/sql/sqlx_indexer_legacy_test.go

@@ -32,7 +32,7 @@ func setupLegacySQLXIndexer(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	ind, err = sql.NewStateDiffIndexer(context.Background(), test.LegacyConfig, db)
+	ind, err = sql.NewStateDiffIndexer(context.Background(), test.LegacyConfig, db, true)
 	require.NoError(t, err)
 }
 

indexer/database/sql/sqlx_indexer_test.go

@@ -34,7 +34,7 @@ func setupSQLXIndexer(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	ind, err = sql.NewStateDiffIndexer(context.Background(), mocks.TestChainConfig, db)
+	ind, err = sql.NewStateDiffIndexer(context.Background(), mocks.TestChainConfig, db, true)
 	require.NoError(t, err)
 }
 

indexer/database/sql/writer.go

@@ -34,13 +34,15 @@ import (
 
 // Writer handles processing and writing of indexed IPLD objects to Postgres
 type Writer struct {
-	db Database
+	db     Database
+	isDiff bool
 }
 
-// NewWriter creates a new pointer to a Writer
-func NewWriter(db Database) *Writer {
+// NewWriter creates a new pointer to a Writer. `diff` indicates whether this is part of an
+// incremental diff (as opposed to a snapshot).
+func NewWriter(db Database, diff bool) *Writer {
 	return &Writer{
-		db: db,
+		db: db, isDiff: diff,
 	}
 }
 
@@ -308,7 +310,8 @@ func (w *Writer) upsertStateCID(tx Tx, stateNode models.StateNodeModel) error {
 		_, err = tx.CopyFrom(w.db.Context(),
 			schema.TableStateNode.TableName(), schema.TableStateNode.ColumnNames(),
 			toRows(toRow(blockNum, stateNode.HeaderID, stateNode.StateKey, stateNode.CID,
-				true, balance, stateNode.Nonce, stateNode.CodeHash, stateNode.StorageRoot, stateNode.Removed)))
+				w.isDiff, balance, stateNode.Nonce, stateNode.CodeHash, stateNode.StorageRoot,
+				stateNode.Removed)))
 		if err != nil {
 			return insertError{"eth.state_cids", err, "COPY", stateNode}
 		}
@@ -318,7 +321,7 @@ func (w *Writer) upsertStateCID(tx Tx, stateNode models.StateNodeModel) error {
 			stateNode.HeaderID,
 			stateNode.StateKey,
 			stateNode.CID,
-			true,
+			w.isDiff,
 			bal,
 			stateNode.Nonce,
 			stateNode.CodeHash,
@@ -346,7 +349,7 @@ func (w *Writer) upsertStorageCID(tx Tx, storageCID models.StorageNodeModel) err
 		_, err = tx.CopyFrom(w.db.Context(),
 			schema.TableStorageNode.TableName(), schema.TableStorageNode.ColumnNames(),
 			toRows(toRow(blockNum, storageCID.HeaderID, storageCID.StateKey, storageCID.StorageKey, storageCID.CID,
-				true, storageCID.Value, storageCID.Removed)))
+				w.isDiff, storageCID.Value, storageCID.Removed)))
 		if err != nil {
 			return insertError{"eth.storage_cids", err, "COPY", storageCID}
 		}
@@ -357,7 +360,7 @@ func (w *Writer) upsertStorageCID(tx Tx, storageCID models.StorageNodeModel) err
 			storageCID.StateKey,
 			storageCID.StorageKey,
 			storageCID.CID,
-			true,
+			w.isDiff,
 			storageCID.Value,
 			storageCID.Removed,
 		)

main/main.go

@@ -52,6 +52,7 @@ func InitializeNode(stack core.Node, b core.Backend) {
 			adapt.ChainConfig(backend.ChainConfig()),
 			info,
 			serviceConfig.IndexerConfig,
+			true,
 		)
 		if err != nil {
 			log.Error("failed to construct indexer", "error", err)