Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feature/DB repair on corruption #4911

Merged
merged 5 commits into from
Nov 17, 2022
Merged

Feature/DB repair on corruption #4911

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a97f765
Added db corruption repair
asdacap Nov 15, 2022
8150a75
Minor cleanup
asdacap Nov 15, 2022
440ac13
Fix whitespace
asdacap Nov 15, 2022
fdf8456
Added more log
asdacap Nov 15, 2022
646d3b6
Addressing comment
asdacap Nov 16, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
248 changes: 217 additions & 31 deletions src/Nethermind/Nethermind.Db.Rocks/DbOnTheRocks.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.IO.Abstractions;
using System.Reflection;
using System.Threading;
using ConcurrentCollections;
Expand Down Expand Up @@ -59,6 +60,14 @@ public class DbOnTheRocks : IDbWithSpan

private readonly RocksDbSettings _settings;

private readonly IFileSystem _fileSystem;

private readonly Func<string, (DbOptions Options, ColumnFamilies? Families), RocksDb> _rocksDbFactory;

private readonly RocksDbSharp.Native _rocksDbNative;

private string CorruptMarkerPath => Path.Join(_fullPath, "corrupt.marker");

protected static void InitCache(IDbConfig dbConfig)
{
if (Interlocked.CompareExchange(ref _cacheInitialized, 1, 0) == 0)
Expand All @@ -68,24 +77,41 @@ protected static void InitCache(IDbConfig dbConfig)
}
}

public DbOnTheRocks(string basePath, RocksDbSettings rocksDbSettings, IDbConfig dbConfig,
ILogManager logManager, ColumnFamilies? columnFamilies = null)
public DbOnTheRocks(
string basePath,
RocksDbSettings rocksDbSettings,
IDbConfig dbConfig,
ILogManager logManager,
ColumnFamilies? columnFamilies = null,
Func<string, (DbOptions Options, ColumnFamilies? Families), RocksDb>? rocksDbFactory = null,
RocksDbSharp.Native? rocksDbNative = null,
IFileSystem? fileSystem = null)
{
_logger = logManager.GetClassLogger();
_settings = rocksDbSettings;
Name = _settings.DbName;
_fileSystem = fileSystem ?? new FileSystem();
_rocksDbFactory = rocksDbFactory ?? DefaultFactory;
_rocksDbNative = rocksDbNative ?? RocksDbSharp.Native.Instance;
_db = Init(basePath, rocksDbSettings.DbPath, dbConfig, logManager, columnFamilies, rocksDbSettings.DeleteOnStart);
}

private static RocksDb DefaultFactory(string path, (DbOptions Options, ColumnFamilies? Families) db)
{
(DbOptions options, ColumnFamilies? families) = db;
return families is null ? RocksDb.Open(options, path) : RocksDb.Open(options, path, families);
}

private RocksDb Open(string path, (DbOptions Options, ColumnFamilies? Families) db)
{
RepairIfCorrupted(db.Options);

return _rocksDbFactory.Invoke(path, db);
}

private RocksDb Init(string basePath, string dbPath, IDbConfig dbConfig, ILogManager? logManager,
ColumnFamilies? columnFamilies = null, bool deleteOnStart = false)
{
static RocksDb Open(string path, (DbOptions Options, ColumnFamilies? Families) db)
{
(DbOptions options, ColumnFamilies? families) = db;
return families is null ? RocksDb.Open(options, path) : RocksDb.Open(options, path, families);
}

_fullPath = GetFullDbPath(dbPath, basePath);
_logger = logManager?.GetClassLogger() ?? NullLogger.Instance;
if (!Directory.Exists(_fullPath))
Expand All @@ -106,7 +132,7 @@ static RocksDb Open(string path, (DbOptions Options, ColumnFamilies? Families) d

// ReSharper disable once VirtualMemberCallInConstructor
if (_logger.IsDebug) _logger.Debug($"Loading DB {Name,-13} from {_fullPath} with max memory footprint of {_maxThisDbSize / 1000 / 1000}MB");
RocksDb db = _dbsByPath.GetOrAdd(_fullPath, static (s, tuple) => Open(s, tuple), (DbOptions, columnFamilies));
RocksDb db = _dbsByPath.GetOrAdd(_fullPath, (s, tuple) => Open(s, tuple), (DbOptions, columnFamilies));

if (dbConfig.EnableMetricsUpdater)
{
Expand All @@ -127,6 +153,36 @@ static RocksDb Open(string path, (DbOptions Options, ColumnFamilies? Families) d
if (_logger.IsWarn) _logger.Warn("If your database did not close properly you need to call 'find -type f -name '*LOCK*' -delete' from the databse folder");
throw;
}
catch (RocksDbSharpException x)
{
CreateMarkerIfCorrupt(x);
throw;
}

}

private void CreateMarkerIfCorrupt(RocksDbSharpException rocksDbException)
{
if (rocksDbException.Message.Contains("Corruption:"))
{
if (_logger.IsWarn) _logger.Warn($"Corrupted DB detected on path {_fullPath}. Please restart Nethermind to attempt repair.");
_fileSystem.File.WriteAllText(CorruptMarkerPath, "marker");
}
}

private void RepairIfCorrupted(DbOptions dbOptions)
{
string corruptMarker = CorruptMarkerPath;

if (!_fileSystem.File.Exists(corruptMarker))
{
return;
}

if (_logger.IsWarn) _logger.Warn($"Corrupted DB marker detected for db {_fullPath}. Attempting repair...");
_rocksDbNative.rocksdb_repair_db(dbOptions.Handle, _fullPath);

_fileSystem.File.Delete(corruptMarker);
}

protected internal void UpdateReadMetrics()
Expand Down Expand Up @@ -280,7 +336,16 @@ public byte[]? this[byte[] key]
}

UpdateReadMetrics();
return _db.Get(key);

try
{
return _db.Get(key);
}
catch (RocksDbSharpException e)
{
CreateMarkerIfCorrupt(e);
throw;
}
}
set
{
Expand All @@ -291,18 +356,40 @@ public byte[]? this[byte[] key]

UpdateWriteMetrics();

if (value is null)
try
{
_db.Remove(key, null, WriteOptions);
if (value is null)
{
_db.Remove(key, null, WriteOptions);
}
else
{
_db.Put(key, value, null, WriteOptions);
}
}
else
catch (RocksDbSharpException e)
{
_db.Put(key, value, null, WriteOptions);
CreateMarkerIfCorrupt(e);
throw;
}
}
}

public KeyValuePair<byte[], byte[]?>[] this[byte[][] keys] => _db.MultiGet(keys);
public KeyValuePair<byte[], byte[]?>[] this[byte[][] keys]
{
get
{
try
{
return _db.MultiGet(keys);
}
catch (RocksDbSharpException e)
{
CreateMarkerIfCorrupt(e);
throw;
}
}
}

public Span<byte> GetSpan(byte[] key)
{
Expand All @@ -313,7 +400,15 @@ public Span<byte> GetSpan(byte[] key)

UpdateReadMetrics();

return _db.GetSpan(key);
try
{
return _db.GetSpan(key);
}
catch (RocksDbSharpException e)
{
CreateMarkerIfCorrupt(e);
throw;
}
}

public void DangerousReleaseMemory(in Span<byte> span) => _db.DangerousReleaseMemory(span);
Expand All @@ -325,7 +420,15 @@ public void Remove(byte[] key)
throw new ObjectDisposedException($"Attempted to delete form a disposed database {Name}");
}

_db.Remove(key, null, WriteOptions);
try
{
_db.Remove(key, null, WriteOptions);
}
catch (RocksDbSharpException e)
{
CreateMarkerIfCorrupt(e);
throw;
}
}

public IEnumerable<KeyValuePair<byte[], byte[]>> GetAll(bool ordered = false)
Expand All @@ -343,7 +446,16 @@ protected internal Iterator CreateIterator(bool ordered = false, ColumnFamilyHan
{
ReadOptions readOptions = new();
readOptions.SetTailing(!ordered);
return _db.NewIterator(ch, readOptions);

try
{
return _db.NewIterator(ch, readOptions);
}
catch (RocksDbSharpException e)
{
CreateMarkerIfCorrupt(e);
throw;
}
}

public IEnumerable<byte[]> GetAllValues(bool ordered = false)
Expand All @@ -359,14 +471,39 @@ public IEnumerable<byte[]> GetAllValues(bool ordered = false)

internal IEnumerable<byte[]> GetAllValuesCore(Iterator iterator)
{
iterator.SeekToFirst();
try
{
iterator.SeekToFirst();
}
catch (RocksDbSharpException e)
{
CreateMarkerIfCorrupt(e);
throw;
}

while (iterator.Valid())
{
yield return iterator.Value();
iterator.Next();
try
{
iterator.Next();
}
catch (RocksDbSharpException e)
{
CreateMarkerIfCorrupt(e);
throw;
}
}

iterator.Dispose();
try
{
iterator.Dispose();
}
catch (RocksDbSharpException e)
{
CreateMarkerIfCorrupt(e);
throw;
}
}

public IEnumerable<KeyValuePair<byte[], byte[]>> GetAllCore(Iterator iterator)
Expand All @@ -376,14 +513,40 @@ public IEnumerable<KeyValuePair<byte[], byte[]>> GetAllCore(Iterator iterator)
throw new ObjectDisposedException($"Attempted to read form a disposed database {Name}");
}

iterator.SeekToFirst();
try
{
iterator.SeekToFirst();
}
catch (RocksDbSharpException e)
{
CreateMarkerIfCorrupt(e);
throw;
}

while (iterator.Valid())
{
yield return new KeyValuePair<byte[], byte[]>(iterator.Key(), iterator.Value());
iterator.Next();

try
{
iterator.Next();
}
catch (RocksDbSharpException e)
{
CreateMarkerIfCorrupt(e);
throw;
}
}

iterator.Dispose();
try
{
iterator.Dispose();
}
catch (RocksDbSharpException e)
{
CreateMarkerIfCorrupt(e);
throw;
}
}

public bool KeyExists(byte[] key)
Expand All @@ -393,9 +556,17 @@ public bool KeyExists(byte[] key)
throw new ObjectDisposedException($"Attempted to read form a disposed database {Name}");
}

// seems it has no performance impact
return _db.Get(key) is not null;
// return _db.Get(key, 32, _keyExistsBuffer, 0, 0, null, null) != -1;
try
{
// seems it has no performance impact
return _db.Get(key) is not null;
// return _db.Get(key, 32, _keyExistsBuffer, 0, 0, null, null) != -1;
}
catch (RocksDbSharpException e)
{
CreateMarkerIfCorrupt(e);
throw;
}
}

public IBatch StartBatch()
Expand Down Expand Up @@ -437,9 +608,17 @@ public void Dispose()
}
_isDisposed = true;

_dbOnTheRocks._db.Write(_rocksBatch, _dbOnTheRocks.WriteOptions);
_dbOnTheRocks._currentBatches.TryRemove(this);
_rocksBatch.Dispose();
try
{
_dbOnTheRocks._db.Write(_rocksBatch, _dbOnTheRocks.WriteOptions);
_dbOnTheRocks._currentBatches.TryRemove(this);
_rocksBatch.Dispose();
}
catch (RocksDbSharpException e)
{
_dbOnTheRocks.CreateMarkerIfCorrupt(e);
throw;
}
}

public byte[]? this[byte[] key]
Expand Down Expand Up @@ -480,7 +659,14 @@ public void Flush()

private void InnerFlush()
{
RocksDbSharp.Native.Instance.rocksdb_flush(_db.Handle, FlushOptions.DefaultFlushOptions.Handle);
try
{
RocksDbSharp.Native.Instance.rocksdb_flush(_db.Handle, FlushOptions.DefaultFlushOptions.Handle);
}
catch (RocksDbSharpException e)
{
CreateMarkerIfCorrupt(e);
}
}

public void Clear()
Expand Down
Loading