diff --git a/kv/kv_interface.go b/kv/kv_interface.go index 12d1003bd..0b0545104 100644 --- a/kv/kv_interface.go +++ b/kv/kv_interface.go @@ -264,6 +264,7 @@ type RwTx interface { // CollectMetrics - does collect all DB-related and Tx-related metrics // this method exists only in RwTx to avoid concurrency CollectMetrics() + Reset() error } // BucketMigrator used for buckets migration, don't use it in usual app code diff --git a/kv/mdbx/kv_mdbx.go b/kv/mdbx/kv_mdbx.go index 1f97cc137..84bd68d14 100644 --- a/kv/mdbx/kv_mdbx.go +++ b/kv/mdbx/kv_mdbx.go @@ -1046,6 +1046,28 @@ func (tx *MdbxTx) DBSize() (uint64, error) { return info.Geo.Current, err } +func (tx *MdbxTx) Reset() (err error) { + tx.Rollback() + //tx.printDebugInfo() + if tx.db.closed.Load() { + return fmt.Errorf("db closed") + } + runtime.LockOSThread() + defer func() { + if err == nil { + tx.db.wg.Add(1) + } + }() + + tx.tx, err = tx.db.env.BeginTxn(nil, 0) + if err != nil { + runtime.UnlockOSThread() // unlock only in case of error. normal flow is "defer .Rollback()" + return fmt.Errorf("%w, lable: %s, trace: %s", err, tx.db.opts.label.String(), stack2.Trace().String()) + } + tx.tx.RawRead = true + return nil +} + func (tx *MdbxTx) RwCursor(bucket string) (kv.RwCursor, error) { b := tx.db.buckets[bucket] if b.AutoDupSortKeysConversion { diff --git a/kv/memdb/memory_mutation.go b/kv/memdb/memory_mutation.go index cc3416a73..8d9c701e6 100644 --- a/kv/memdb/memory_mutation.go +++ b/kv/memdb/memory_mutation.go @@ -374,7 +374,10 @@ func (m *MemoryMutation) Cursor(bucket string) (kv.Cursor, error) { return m.makeCursor(bucket) } -// ViewID creates a new cursor (the real fun begins here) func (m *MemoryMutation) ViewID() uint64 { panic("ViewID Not implemented") } + +func (m *MemoryMutation) Reset() error { + return nil +}