NODE-2622 Delete old data (#3912)

wavesplatform · Jan 16, 2024 · f3cb220 · f3cb220
1 parent 247af02
commit f3cb220
Show file tree

Hide file tree

Showing 42 changed files with 918 additions and 344 deletions.
diff --git a/benchmark/src/main/scala/com/wavesplatform/state/DBState.scala b/benchmark/src/main/scala/com/wavesplatform/state/DBState.scala
@@ -22,13 +22,12 @@ abstract class DBState extends ScorexLogging {
 
   lazy val rdb: RDB = RDB.open(settings.dbSettings)
 
-  lazy val rocksDBWriter: RocksDBWriter =
-    new RocksDBWriter(
-      rdb,
-      settings.blockchainSettings,
-      settings.dbSettings.copy(maxCacheSize = 1),
-      settings.enableLightMode
-    )
+  lazy val rocksDBWriter: RocksDBWriter = RocksDBWriter(
+    rdb,
+    settings.blockchainSettings,
+    settings.dbSettings.copy(maxCacheSize = 1),
+    settings.enableLightMode
+  )
 
   AddressScheme.current = new AddressScheme { override val chainId: Byte = 'W' }
 
@@ -44,6 +43,7 @@ abstract class DBState extends ScorexLogging {
 
   @TearDown
   def close(): Unit = {
+    rocksDBWriter.close()
     rdb.close()
   }
 }
diff --git a/benchmark/src/test/scala/com/wavesplatform/RollbackBenchmark.scala b/benchmark/src/test/scala/com/wavesplatform/RollbackBenchmark.scala
@@ -22,7 +22,7 @@ object RollbackBenchmark extends ScorexLogging {
     val settings      = Application.loadApplicationConfig(Some(new File(args(0))))
     val rdb           = RDB.open(settings.dbSettings)
     val time          = new NTP(settings.ntpServer)
-    val rocksDBWriter = new RocksDBWriter(rdb, settings.blockchainSettings, settings.dbSettings, settings.enableLightMode)
+    val rocksDBWriter = RocksDBWriter(rdb, settings.blockchainSettings, settings.dbSettings, settings.enableLightMode)
 
     val issuer = KeyPair(new Array[Byte](32))
 
@@ -111,6 +111,7 @@ object RollbackBenchmark extends ScorexLogging {
     rocksDBWriter.rollbackTo(1)
     val end = System.nanoTime()
     log.info(f"Rollback took ${(end - start) * 1e-6}%.3f ms")
+    rocksDBWriter.close()
     rdb.close()
   }
 }
diff --git a/benchmark/src/test/scala/com/wavesplatform/state/BaseState.scala b/benchmark/src/test/scala/com/wavesplatform/state/BaseState.scala
@@ -104,6 +104,7 @@ trait BaseState {
 
   @TearDown
   def close(): Unit = {
+    state.close()
     rdb.close()
   }
 }

diff --git a/benchmark/src/test/scala/com/wavesplatform/state/RocksDBIteratorBenchmark.scala b/benchmark/src/test/scala/com/wavesplatform/state/RocksDBIteratorBenchmark.scala
@@ -1,8 +1,5 @@
 package com.wavesplatform.state
 
-import java.nio.file.Files
-import java.util.concurrent.TimeUnit
-
 import com.google.common.primitives.Ints
 import com.typesafe.config.ConfigFactory
 import com.wavesplatform.database.RDB
@@ -12,6 +9,10 @@ import org.openjdk.jmh.annotations.*
 import org.openjdk.jmh.infra.Blackhole
 import org.rocksdb.{ReadOptions, WriteBatch, WriteOptions}
 
+import java.nio.file.Files
+import java.util.concurrent.TimeUnit
+import scala.util.Using
+
 @OutputTimeUnit(TimeUnit.NANOSECONDS)
 @BenchmarkMode(Array(Mode.AverageTime))
 @Threads(1)
@@ -60,7 +61,7 @@ object RocksDBIteratorBenchmark {
       RDB.open(wavesSettings.dbSettings.copy(directory = dir))
     }
 
-    val keysPrefix            = "keysPrefix"
+    val keysPrefix            = "keysPrefix" // Must have 10 or more bytes, see RDB.newColumnFamilyOptions
     val firstKey: Array[Byte] = keysPrefix.getBytes ++ Ints.toByteArray(1)
     val lastKey: Array[Byte]  = keysPrefix.getBytes ++ Ints.toByteArray(10000)
 
@@ -70,14 +71,18 @@ object RocksDBIteratorBenchmark {
 
     val readOptions: ReadOptions = new ReadOptions().setTotalOrderSeek(false).setPrefixSameAsStart(true)
 
-    private val wb: WriteBatch = new WriteBatch()
-    kvs.foreach { case (key, value) =>
-      wb.put(key, value)
+    Using.Manager { use =>
+      val wb = use(new WriteBatch())
+      val wo = use(new WriteOptions())
+      kvs.foreach { case (key, value) =>
+        wb.put(key, value)
+      }
+      rdb.db.write(wo, wb)
     }
-    rdb.db.write(new WriteOptions(), wb)
 
     @TearDown
     def close(): Unit = {
+      readOptions.close()
       rdb.close()
     }
   }

diff --git a/benchmark/src/test/scala/com/wavesplatform/state/RocksDBSeekForPrevBenchmark.scala b/benchmark/src/test/scala/com/wavesplatform/state/RocksDBSeekForPrevBenchmark.scala
@@ -1,11 +1,7 @@
 package com.wavesplatform.state
 
-import java.nio.file.Files
-import java.util.concurrent.TimeUnit
-
 import com.google.common.primitives.{Bytes, Shorts}
 import com.typesafe.config.ConfigFactory
-import com.wavesplatform.account.Address
 import com.wavesplatform.database.{
   AddressId,
   CurrentData,
@@ -24,6 +20,10 @@ import org.openjdk.jmh.annotations.*
 import org.openjdk.jmh.infra.Blackhole
 import org.rocksdb.{ReadOptions, WriteBatch, WriteOptions}
 
+import java.nio.file.Files
+import java.util.concurrent.TimeUnit
+import scala.util.Using
+
 @OutputTimeUnit(TimeUnit.NANOSECONDS)
 @BenchmarkMode(Array(Mode.AverageTime))
 @Threads(1)
@@ -63,27 +63,29 @@ object RocksDBSeekForPrevBenchmark {
       RDB.open(wavesSettings.dbSettings.copy(directory = dir))
     }
 
-    val address: Address     = Address(Array.fill(20)(1.toByte))
     val addressId: AddressId = AddressId(1L)
 
     val keyString                          = "key"
-    val currentDataKey: Array[Byte]        = Keys.data(address, keyString).keyBytes
+    val currentDataKey: Array[Byte]        = Keys.data(addressId, keyString).keyBytes
     val dataNodeKey: Height => Array[Byte] = Keys.dataAt(addressId, "key")(_).keyBytes
     val dataNodeKeyPrefix: Array[Byte] = Bytes.concat(Shorts.toByteArray(KeyTags.DataHistory.id.toShort), addressId.toByteArray, keyString.getBytes)
 
     private val dataEntry: StringDataEntry = StringDataEntry(keyString, "value")
 
     val readOptions: ReadOptions = new ReadOptions()
 
-    private val wb: WriteBatch = new WriteBatch()
-    wb.put(currentDataKey, writeCurrentData(CurrentData(dataEntry, Height(10000), Height(9999))))
-    (1 to 1000).foreach { h =>
-      wb.put(dataNodeKey(Height(h)), writeDataNode(DataNode(dataEntry, Height(h - 1))))
+    Using.Manager { use =>
+      val wb = use(new WriteBatch())
+      wb.put(currentDataKey, writeCurrentData(CurrentData(dataEntry, Height(10000), Height(9999))))
+      (1 to 1000).foreach { h =>
+        wb.put(dataNodeKey(Height(h)), writeDataNode(DataNode(dataEntry, Height(h - 1))))
+      }
+      rdb.db.write(use(new WriteOptions()), wb)
     }
-    rdb.db.write(new WriteOptions(), wb)
 
     @TearDown
     def close(): Unit = {
+      readOptions.close()
       rdb.close()
     }
   }

diff --git a/benchmark/src/test/scala/com/wavesplatform/state/RocksDBWriterBenchmark.scala b/benchmark/src/test/scala/com/wavesplatform/state/RocksDBWriterBenchmark.scala
@@ -1,8 +1,5 @@
 package com.wavesplatform.state
 
-import java.io.File
-import java.util.concurrent.{ThreadLocalRandom, TimeUnit}
-
 import com.typesafe.config.ConfigFactory
 import com.wavesplatform.account.*
 import com.wavesplatform.api.BlockMeta
@@ -17,7 +14,10 @@ import com.wavesplatform.transaction.Transaction
 import org.openjdk.jmh.annotations.*
 import org.openjdk.jmh.infra.Blackhole
 
+import java.io.File
+import java.util.concurrent.{ThreadLocalRandom, TimeUnit}
 import scala.io.Codec
+import scala.util.Using
 
 /** Tests over real database. How to test:
   *   1. Download a database 2. Import it:
@@ -87,7 +87,7 @@ object RocksDBWriterBenchmark {
       RDB.open(wavesSettings.dbSettings)
     }
 
-    val db = new RocksDBWriter(rawDB, wavesSettings.blockchainSettings, wavesSettings.dbSettings, wavesSettings.enableLightMode)
+    val db = RocksDBWriter(rawDB, wavesSettings.blockchainSettings, wavesSettings.dbSettings, wavesSettings.enableLightMode)
 
     def loadBlockInfoAt(height: Int): Option[(BlockMeta, Seq[(TxMeta, Transaction)])] =
       loadBlockMetaAt(height).map { meta =>
@@ -102,15 +102,12 @@ object RocksDBWriterBenchmark {
 
     @TearDown
     def close(): Unit = {
+      db.close()
       rawDB.close()
     }
 
     protected def load[T](label: String, absolutePath: String)(f: String => T): Vector[T] = {
-      scala.io.Source
-        .fromFile(absolutePath)(Codec.UTF8)
-        .getLines()
-        .map(f)
-        .toVector
+      Using.resource(scala.io.Source.fromFile(absolutePath)(Codec.UTF8))(_.getLines().map(f).toVector)
     }
   }
 

diff --git a/benchmark/src/test/scala/com/wavesplatform/state/WavesEnvironmentBenchmark.scala b/benchmark/src/test/scala/com/wavesplatform/state/WavesEnvironmentBenchmark.scala
@@ -21,6 +21,7 @@ import scodec.bits.BitVector
 import java.io.File
 import java.util.concurrent.{ThreadLocalRandom, TimeUnit}
 import scala.io.Codec
+import scala.util.Using
 
 /** Tests over real database. How to test:
   *   1. Download a database 2. Import it:
@@ -134,8 +135,8 @@ object WavesEnvironmentBenchmark {
       RDB.open(wavesSettings.dbSettings)
     }
 
+    val state = RocksDBWriter(rdb, wavesSettings.blockchainSettings, wavesSettings.dbSettings, wavesSettings.enableLightMode)
     val environment: Environment[Id] = {
-      val state = new RocksDBWriter(rdb, wavesSettings.blockchainSettings, wavesSettings.dbSettings, wavesSettings.enableLightMode)
       WavesEnvironment(
         AddressScheme.current.chainId,
         Coeval.raiseError(new NotImplementedError("`tx` is not implemented")),
@@ -149,15 +150,12 @@ object WavesEnvironmentBenchmark {
 
     @TearDown
     def close(): Unit = {
+      state.close()
       rdb.close()
     }
 
     protected def load[T](label: String, absolutePath: String)(f: String => T): Vector[T] = {
-      scala.io.Source
-        .fromFile(absolutePath)(Codec.UTF8)
-        .getLines()
-        .map(f)
-        .toVector
+      Using.resource(scala.io.Source.fromFile(absolutePath)(Codec.UTF8))(_.getLines().map(f).toVector)
     }
   }
 

diff --git a/node-it/src/test/scala/com/wavesplatform/it/BaseTargetChecker.scala b/node-it/src/test/scala/com/wavesplatform/it/BaseTargetChecker.scala
@@ -23,11 +23,11 @@ object BaseTargetChecker {
       .withFallback(defaultReference())
       .resolve()
 
-    val settings               = WavesSettings.fromRootConfig(sharedConfig)
-    val db                     = RDB.open(settings.dbSettings.copy(directory = "/tmp/tmp-db"))
-    val ntpTime                = new NTP("ntp.pool.org")
-    val (blockchainUpdater, _) = StorageFactory(settings, db, ntpTime, BlockchainUpdateTriggers.noop)
-    val poSSelector            = PoSSelector(blockchainUpdater, settings.synchronizationSettings.maxBaseTarget)
+    val settings                       = WavesSettings.fromRootConfig(sharedConfig)
+    val db                             = RDB.open(settings.dbSettings.copy(directory = "/tmp/tmp-db"))
+    val ntpTime                        = new NTP("ntp.pool.org")
+    val (blockchainUpdater, rdbWriter) = StorageFactory(settings, db, ntpTime, BlockchainUpdateTriggers.noop)
+    val poSSelector                    = PoSSelector(blockchainUpdater, settings.synchronizationSettings.maxBaseTarget)
 
     try {
       val genesisBlock =
@@ -51,6 +51,10 @@ object BaseTargetChecker {
 
           f"$address: ${timeDelay * 1e-3}%10.3f s"
       }
-    } finally ntpTime.close()
+    } finally {
+      ntpTime.close()
+      rdbWriter.close()
+      db.close()
+    }
   }
 }
diff --git a/node-it/src/test/scala/com/wavesplatform/it/Docker.scala b/node-it/src/test/scala/com/wavesplatform/it/Docker.scala
@@ -349,6 +349,7 @@ class Docker(
     client.startContainer(id)
     nodes.asScala.find(_.containerId == id).foreach { node =>
       node.nodeInfo = getNodeInfo(node.containerId, node.settings)
+      Await.result(node.waitForStartup(), 3.minutes)
     }
   }