Fix WAL file buildup (#3603)

Due to the wrong CF being passed to the database flush mechanism during
sync, the sync KVT would remain unflushed as a memtable and thus prevent
the WAL from being released - given that a new wal is created for every
flush, this would lead to a buildup of small WAL files (50k and counting
on my hoodi node) after a sync.

At the same time, we don't really need to flush the KVT at every commit
since the skiplist permits efficient lookups (unlike the vector memtable
used for bulk updates to the state).

Author: arnetheduck

execution_chain/db/aristo/aristo_init/rocks_db/rdb_put.nim

@@ -46,7 +46,7 @@ proc rollback*(rdb: var RdbInst, session: SharedWriteBatchRef) =
 proc commit*(rdb: var RdbInst, session: SharedWriteBatchRef): Result[void,(AristoError,string)] =
   if not session.isClosed():
     defer: session.close()
-    rdb.baseDb.commit(session, rdb.vtxCol).isOkOr:
+    rdb.baseDb.commit(session, rdb.vtxCol, true).isOkOr:
       const errSym = RdbBeDriverWriteError
       when extraTraceMessages:
         trace logTxt "commit", error=errSym, info=error

execution_chain/db/core_db/backend/aristo_rocksdb.nim

@@ -42,6 +42,17 @@ proc toDbOpts*(opts: DbOptions): DbOptionsRef =
     # https://github.com/facebook/rocksdb/blob/af50823069818fc127438e39fef91d2486d6e76c/include/rocksdb/options.h#L1276
     dbOpts.rowCache = cacheCreateLRU(opts.rowCacheSize, autoClose = true)
 
+  # The WAL is shared between column families and a single small unflushed CF
+  # (like the sync cache) can keep the WAL growing for a long time - this both
+  # causes poor startup times and results in an ever-growing number of WAL files.
+  # The default value is very large, specially given that each manual flush
+  # of the aristo table creates a new log file.
+  # https://github.com/facebook/rocksdb/blob/af50823069818fc127438e39fef91d2486d6e76c/include/rocksdb/options.h#L719
+  let writeBufferSize =
+    if opts.writeBufferSize > 0: opts.writeBufferSize else: defaultWriteBufferSize
+
+  dbOpts.maxTotalWalSize = 3 * writeBufferSize + 1024 * 1024
+
   dbOpts.keepLogFileNum = 16 # No point keeping 1000 log files around...
 
   # Parallelize L0 -> Ln compaction

execution_chain/db/core_db/backend/rocksdb_desc.nim

@@ -58,13 +58,20 @@ proc close*(session: SharedWriteBatchRef) =
     session.closes = 0
 
 proc commit*(
-    rdb: RocksDbInstanceRef, session: SharedWriteBatchRef, cf: ColFamilyReadWrite
+    rdb: RocksDbInstanceRef,
+    session: SharedWriteBatchRef,
+    cf: ColFamilyReadWrite,
+    flush: bool,
 ): Result[void, string] =
   session.commits += 1
-  session.families.add cf
+
+  if flush:
+    session.families.add cf
+
   if session.commits == session.refs:
     # Write to disk if everyone that opened a session also committed it
     ?rdb.db.write(session.batch)
+
     # This flush forces memtables to be written to disk, which is necessary given
     # the use of vector memtables which have very bad lookup performance.
     rdb.db.flush(session.families.mapIt(it.handle())).isOkOr:
@@ -80,7 +87,6 @@ proc open*(
     dbOpts: DbOptionsRef,
     cfs: openArray[(string, ColFamilyOptionsRef)],
 ): Result[RocksDbInstanceRef, string] =
-
   let ecdbDir = baseDir.ecdbDir
 
   if not dirExists(ecdbDir):

execution_chain/db/kvt/kvt_init/rocks_db.nim

@@ -122,7 +122,7 @@ proc putKvpFn(db: RdbBackendRef, cf: static[KvtCFs]): PutKvpFn =
           return
 
 
-proc putEndFn(db: RdbBackendRef): PutEndFn =
+proc putEndFn(db: RdbBackendRef, cf: static[KvtCFs]): PutEndFn =
   result =
     proc(hdl: PutHdlRef): Result[void,KvtError] =
       let hdl = hdl.endSession db
@@ -133,7 +133,7 @@ proc putEndFn(db: RdbBackendRef): PutEndFn =
         return err(hdl.error)
 
       # Commit session
-      db.rdb.commit(hdl.session).isOkOr:
+      db.rdb.commit(hdl.session, cf).isOkOr:
         when extraTraceMessages:
           trace "putEndFn: failed", error=($error[0]), info=error[1]
           return err(error[0])
@@ -170,7 +170,7 @@ proc rocksDbKvtBackend*(baseDb: RocksDbInstanceRef, cf: static[KvtCFs]): KvtDbRe
 
   db.putBegFn = putBegFn be
   db.putKvpFn = putKvpFn(be, cf)
-  db.putEndFn = putEndFn be
+  db.putEndFn = putEndFn(be, cf)
 
   db.closeFn = closeFn be
   db.getBackendFn = getBackendFn be

execution_chain/db/kvt/kvt_init/rocks_db/rdb_put.nim

@@ -49,10 +49,12 @@ proc rollback*(rdb: var RdbInst, session: SharedWriteBatchRef) =
   if not session.isClosed():
     session.close()
 
-proc commit*(rdb: var RdbInst, session: SharedWriteBatchRef): Result[void,(KvtError,string)] =
+proc commit*(
+    rdb: var RdbInst, session: SharedWriteBatchRef, cf: static[KvtCFs]
+): Result[void, (KvtError, string)] =
   if not session.isClosed():
     defer: session.close()
-    rdb.baseDb.commit(session, rdb.store[KvtGeneric]).isOkOr:
+    rdb.baseDb.commit(session, rdb.store[cf], false).isOkOr:
       const errSym = RdbBeDriverWriteError
       when extraTraceMessages:
         trace logTxt "commit", error=errSym, info=error