[wip] Close cursor if it's opened by different tx

core/rawdb/rawtemporaldb/accessors_receipt_test.go

@@ -20,10 +20,11 @@ func TestAppendReceipt(t *testing.T) {
 	require.NoError(err)
 	defer tx.Rollback()
 
-	doms, err := state.NewSharedDomains(tx, log.New())
+	ttx := tx.(kv.TemporalTx)
+	doms, err := state.NewSharedDomains(ttx, log.New())
 	require.NoError(err)
 	defer doms.Close()
-	doms.SetTx(tx)
+	doms.SetTx(ttx)
 
 	doms.SetTxNum(0) // block1
 	err = AppendReceipt(doms, &types.Receipt{CumulativeGasUsed: 10, FirstLogIndexWithinBlock: 0}, 0)
@@ -48,7 +49,6 @@ func TestAppendReceipt(t *testing.T) {
 	err = doms.Flush(context.Background(), tx)
 	require.NoError(err)
 
-	ttx := tx.(kv.TemporalTx)
 	v, ok, err := ttx.HistorySeek(kv.ReceiptDomain, FirstLogIndexKey, 0)
 	require.NoError(err)
 	require.True(ok)

core/vm/gas_table_test.go

@@ -22,9 +22,11 @@ package vm_test
 import (
 	"context"
 	"errors"
+	"fmt"
 	"math"
 	"strconv"
 	"testing"
+	"unsafe"
 
 	"github.com/holiman/uint256"
 	"github.com/stretchr/testify/require"
@@ -193,12 +195,16 @@ func TestCreateGas(t *testing.T) {
 		var txc wrap.TxContainer
 		txc.Tx = tx
 
-		domains, err := state3.NewSharedDomains(tx, log.New())
+		eface := *(*[2]uintptr)(unsafe.Pointer(&tx))
+		fmt.Printf("init tx %x\n", eface[1])
+
+		domains, err := state3.NewSharedDomains(txc.Tx, log.New())
 		require.NoError(t, err)
 		defer domains.Close()
 		txc.Doms = domains
 
-		stateReader = rpchelper.NewLatestStateReader(tx)
+		//stateReader = rpchelper.NewLatestStateReader(domains)
+		stateReader = rpchelper.NewLatestDomainStateReader(domains)
 		stateWriter = rpchelper.NewLatestStateWriter(txc, nil, 0)
 
 		s := state.New(stateReader)
@@ -226,5 +232,6 @@ func TestCreateGas(t *testing.T) {
 			t.Errorf("test %d: gas used mismatch: have %v, want %v", i, gasUsed, tt.gasUsed)
 		}
 		tx.Rollback()
+		domains.Close()
 	}
 }

erigon-lib/state/aggregator_test.go

@@ -52,6 +52,215 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestAggregatorV3_Merge(t *testing.T) {
+	t.Parallel()
+	db, agg := testDbAndAggregatorv3(t, 10)
+	rwTx, err := db.BeginRwNosync(context.Background())
+	require.NoError(t, err)
+	defer func() {
+		if rwTx != nil {
+			rwTx.Rollback()
+		}
+	}()
+
+	ac := agg.BeginFilesRo()
+	defer ac.Close()
+	domains, err := NewSharedDomains(WrapTxWithCtx(rwTx, ac), log.New())
+	require.NoError(t, err)
+	defer domains.Close()
+
+	txs := uint64(1000)
+	rnd := rand.New(rand.NewSource(time.Now().UnixNano()))
+
+	var (
+		commKey1 = []byte("someCommKey")
+		commKey2 = []byte("otherCommKey")
+	)
+
+	// keys are encodings of numbers 1..31
+	// each key changes value on every txNum which is multiple of the key
+	var maxWrite, otherMaxWrite uint64
+	for txNum := uint64(1); txNum <= txs; txNum++ {
+		domains.SetTxNum(txNum)
+
+		addr, loc := make([]byte, length.Addr), make([]byte, length.Hash)
+
+		n, err := rnd.Read(addr)
+		require.NoError(t, err)
+		require.EqualValues(t, length.Addr, n)
+
+		n, err = rnd.Read(loc)
+		require.NoError(t, err)
+		require.EqualValues(t, length.Hash, n)
+
+		buf := types.EncodeAccountBytesV3(1, uint256.NewInt(0), nil, 0)
+		err = domains.DomainPut(kv.AccountsDomain, addr, nil, buf, nil, 0)
+		require.NoError(t, err)
+
+		err = domains.DomainPut(kv.StorageDomain, addr, loc, []byte{addr[0], loc[0]}, nil, 0)
+		require.NoError(t, err)
+
+		var v [8]byte
+		binary.BigEndian.PutUint64(v[:], txNum)
+		if txNum%135 == 0 {
+			// pv, step, _, err := ac.GetLatest(kv.CommitmentDomain, commKey2, nil, rwTx)
+			pv, step, err := domains.DomainGet(kv.CommitmentDomain, commKey2, nil)
+			require.NoError(t, err)
+
+			err = domains.DomainPut(kv.CommitmentDomain, commKey2, nil, v[:], pv, step)
+			require.NoError(t, err)
+			otherMaxWrite = txNum
+		} else {
+			// pv, step, _, err := ac.GetLatest(kv.CommitmentDomain, commKey1, nil, rwTx)
+			pv, step, err := domains.DomainGet(kv.CommitmentDomain, commKey1, nil)
+			require.NoError(t, err)
+
+			err = domains.DomainPut(kv.CommitmentDomain, commKey1, nil, v[:], pv, step)
+			require.NoError(t, err)
+			maxWrite = txNum
+		}
+		require.NoError(t, err)
+
+	}
+
+	err = domains.Flush(context.Background(), rwTx)
+	require.NoError(t, err)
+
+	require.NoError(t, err)
+	err = rwTx.Commit()
+	require.NoError(t, err)
+	rwTx = nil
+
+	err = agg.BuildFiles(txs)
+	require.NoError(t, err)
+
+	rwTx, err = db.BeginRw(context.Background())
+	require.NoError(t, err)
+	defer rwTx.Rollback()
+
+	logEvery := time.NewTicker(30 * time.Second)
+	defer logEvery.Stop()
+	stat, err := ac.Prune(context.Background(), rwTx, 0, logEvery)
+	require.NoError(t, err)
+	t.Logf("Prune: %s", stat)
+
+	err = rwTx.Commit()
+	require.NoError(t, err)
+
+	err = agg.MergeLoop(context.Background())
+	require.NoError(t, err)
+
+	// Check the history
+	roTx, err := db.BeginRo(context.Background())
+	require.NoError(t, err)
+	defer roTx.Rollback()
+
+	dc := agg.BeginFilesRo()
+
+	v, _, ex, err := dc.GetLatest(kv.CommitmentDomain, commKey1, nil, roTx)
+	require.NoError(t, err)
+	require.Truef(t, ex, "key %x not found", commKey1)
+
+	require.EqualValues(t, maxWrite, binary.BigEndian.Uint64(v[:]))
+
+	v, _, ex, err = dc.GetLatest(kv.CommitmentDomain, commKey2, nil, roTx)
+	require.NoError(t, err)
+	require.Truef(t, ex, "key %x not found", commKey2)
+	dc.Close()
+
+	require.EqualValues(t, otherMaxWrite, binary.BigEndian.Uint64(v[:]))
+}
+
+func TestAggregatorV3_MergeValTransform(t *testing.T) {
+	t.Parallel()
+	db, agg := testDbAndAggregatorv3(t, 10)
+	rwTx, err := db.BeginRwNosync(context.Background())
+	require.NoError(t, err)
+	defer func() {
+		if rwTx != nil {
+			rwTx.Rollback()
+		}
+	}()
+	ac := agg.BeginFilesRo()
+	defer ac.Close()
+	domains, err := NewSharedDomains(WrapTxWithCtx(rwTx, ac), log.New())
+	require.NoError(t, err)
+	defer domains.Close()
+
+	txs := uint64(1000)
+	rnd := rand.New(rand.NewSource(time.Now().UnixNano()))
+
+	agg.commitmentValuesTransform = true
+
+	state := make(map[string][]byte)
+
+	// keys are encodings of numbers 1..31
+	// each key changes value on every txNum which is multiple of the key
+	//var maxWrite, otherMaxWrite uint64
+	for txNum := uint64(1); txNum <= txs; txNum++ {
+		domains.SetTxNum(txNum)
+
+		addr, loc := make([]byte, length.Addr), make([]byte, length.Hash)
+
+		n, err := rnd.Read(addr)
+		require.NoError(t, err)
+		require.EqualValues(t, length.Addr, n)
+
+		n, err = rnd.Read(loc)
+		require.NoError(t, err)
+		require.EqualValues(t, length.Hash, n)
+
+		buf := types.EncodeAccountBytesV3(1, uint256.NewInt(txNum*1e6), nil, 0)
+		err = domains.DomainPut(kv.AccountsDomain, addr, nil, buf, nil, 0)
+		require.NoError(t, err)
+
+		err = domains.DomainPut(kv.StorageDomain, addr, loc, []byte{addr[0], loc[0]}, nil, 0)
+		require.NoError(t, err)
+
+		if (txNum+1)%agg.StepSize() == 0 {
+			_, err := domains.ComputeCommitment(context.Background(), true, txNum/10, "")
+			require.NoError(t, err)
+		}
+
+		state[string(addr)] = buf
+		state[string(addr)+string(loc)] = []byte{addr[0], loc[0]}
+	}
+
+	err = domains.Flush(context.Background(), rwTx)
+	require.NoError(t, err)
+
+	err = rwTx.Commit()
+	require.NoError(t, err)
+	rwTx = nil
+
+	err = agg.BuildFiles(txs)
+	require.NoError(t, err)
+
+	ac.Close()
+	ac = agg.BeginFilesRo()
+	defer ac.Close()
+
+	rwTx, err = db.BeginRwNosync(context.Background())
+	require.NoError(t, err)
+	defer func() {
+		if rwTx != nil {
+			rwTx.Rollback()
+		}
+	}()
+
+	logEvery := time.NewTicker(30 * time.Second)
+	defer logEvery.Stop()
+	stat, err := ac.Prune(context.Background(), rwTx, 0, logEvery)
+	require.NoError(t, err)
+	t.Logf("Prune: %s", stat)
+
+	err = rwTx.Commit()
+	require.NoError(t, err)
+
+	err = agg.MergeLoop(context.Background())
+	require.NoError(t, err)
+}
+
 func TestAggregatorV3_RestartOnDatadir(t *testing.T) {
 	t.Parallel()
 	//t.Skip()

erigon-lib/state/domain.go

@@ -26,7 +26,9 @@ import (
 	"path/filepath"
 	"sort"
 	"sync"
+	"sync/atomic"
 	"time"
+	"unsafe"
 
 	"github.com/erigontech/erigon-lib/metrics"
 	btree2 "github.com/tidwall/btree"
@@ -654,7 +656,8 @@ type DomainRoTx struct {
 	keyBuf [60]byte // 52b key and 8b for inverted step
 	comBuf []byte
 
-	valsC kv.Cursor
+	valsC       kv.Cursor
+	vcParentPtr atomic.Uintptr
 
 	getFromFileCache *DomainGetFromFileCache
 }
@@ -1648,6 +1651,7 @@ func (dt *DomainRoTx) Close() {
 	if dt.files == nil { // invariant: it's safe to call Close multiple times
 		return
 	}
+	dt.closeValsCursor()
 	files := dt.files
 	dt.files = nil
 	for i := range files {
@@ -1715,10 +1719,39 @@ func (dt *DomainRoTx) statelessBtree(i int) *BtIndex {
 	return r
 }
 
-func (dt *DomainRoTx) valsCursor(tx kv.Tx) (c kv.Cursor, err error) {
+var sdTxImmutabilityInvariant = errors.New("tx passed into ShredDomains is immutable")
+
+func (dt *DomainRoTx) closeValsCursor() {
 	if dt.valsC != nil {
+		dt.valsC.Close()
+		dt.vcParentPtr.Store(0)
+		dt.valsC = nil
+	}
+}
+func (dt *DomainRoTx) valsCursor(tx kv.Tx) (c kv.Cursor, err error) {
+	eface := *(*[2]uintptr)(unsafe.Pointer(&tx))
+	if dt.valsC != nil { // run in assert mode only
+		fmt.Printf("cmp e1=%x e2=%x s=%x d=%s\n", eface[0], eface[1], dt.vcParentPtr.Load(), dt.d.filenameBase)
+		if !dt.vcParentPtr.CompareAndSwap(eface[1], eface[1]) { // cant swap when parent ptr is different
+			//if sd, ok := tx.(kv.TemporalPutDel); ok {
+			//	if stx, ok := sd.(*SharedDomains); ok {
+			//		ttx := stx.Tx()
+			//		eface := *(*[2]uintptr)(unsafe.Pointer(&ttx))
+			//		if !dt.vcParentPtr.CompareAndSwap(eface[1], eface[1]) { // cant swap when parent ptr is different
+			//			panic(fmt.Errorf("%w: cursor parent tx %x; current sd tx %x", sdTxImmutabilityInvariant, dt.vcParentPtr.Load(), eface[1])) // cursor opened by different tx, invariant broken
+			//		}
+			//	}
+			//} else {
+			panic(fmt.Errorf("%w: cursor parent tx %x; current tx %x", sdTxImmutabilityInvariant, dt.vcParentPtr.Load(), eface[1])) // cursor opened by different tx, invariant broken
+			//}
+		}
 		return dt.valsC, nil
 	}
+	// initialise parent pointer tracking
+	if !dt.vcParentPtr.CompareAndSwap(0, eface[1]) {
+		panic(fmt.Errorf("%w: cursor parent tx %x; current tx %x", sdTxImmutabilityInvariant, dt.vcParentPtr.Load(), eface[1])) // cursor opened by different tx, invariant broken
+	}
+	fmt.Printf("set e1=%x e2=%x d=%s\n", eface[0], eface[1], dt.d.filenameBase)
 
 	if dt.d.largeVals {
 		dt.valsC, err = tx.Cursor(dt.d.valsTable)

erigon-lib/state/domain_shared.go

@@ -915,13 +915,14 @@ func (sd *SharedDomains) Flush(ctx context.Context, tx kv.RwTx) error {
 		_, f, l, _ := runtime.Caller(1)
 		fmt.Printf("[SD aggTx=%d] FLUSHING at tx %d [%x], caller %s:%d\n", sd.aggTx.id, sd.TxNum(), fh, filepath.Base(f), l)
 	}
-	for _, w := range sd.domainWriters {
+	for di, w := range sd.domainWriters {
 		if w == nil {
 			continue
 		}
 		if err := w.Flush(ctx, tx); err != nil {
 			return err
 		}
+		sd.aggTx.d[di].closeValsCursor()
 	}
 	for _, w := range sd.iiWriters {
 		if w == nil {