more plumbing (old Scavenge done)

Author: timothycoleman

src/EventStore.Core.Tests/Services/Storage/ReadIndexTestScenario.cs

@@ -153,7 +153,8 @@ public override async Task TestFixtureSetUp() {
 			if (_scavenge) {
 				if (_completeLastChunkOnScavenge)
 					Db.Manager.GetChunk(Db.Manager.ChunksCount - 1).Complete();
-				_scavenger = new TFChunkScavenger<TStreamId>(Serilog.Log.Logger, Db, new FakeTFScavengerLog(), TableIndex, ReadIndex, _logFormat.Metastreams);
+				_scavenger = new TFChunkScavenger<TStreamId>(Serilog.Log.Logger, Db, new FakeTFScavengerLog(), TableIndex, ReadIndex, _logFormat.Metastreams,
+					ITransactionFileTrackerFactory.NoOp);
 				await _scavenger.Scavenge(alwaysKeepScavenged: true, mergeChunks: _mergeChunks,
 					scavengeIndex: _scavengeIndex);
 			}

src/EventStore.Core.Tests/TransactionLog/Scavenging/Helpers/ScavengeLifeCycleScenario.cs

@@ -2,6 +2,7 @@
 using System.Threading.Tasks;
 using EventStore.Core.LogAbstraction;
 using EventStore.Core.Tests.Services.Storage;
+using EventStore.Core.TransactionLog;
 using EventStore.Core.TransactionLog.Chunks;
 using NUnit.Framework;
 
@@ -42,7 +43,7 @@ public override async Task TestFixtureSetUp() {
 			Log = new FakeTFScavengerLog();
 			FakeTableIndex = new FakeTableIndex<TStreamId>();
 			TfChunkScavenger = new TFChunkScavenger<TStreamId>(Serilog.Log.Logger, _dbResult.Db, Log, FakeTableIndex, new FakeReadIndex<TLogFormat, TStreamId>(_ => false, _logFormat.Metastreams),
-				_logFormat.Metastreams);
+				_logFormat.Metastreams, ITransactionFileTrackerFactory.NoOp);
 
 			try {
 				await When().WithTimeout(TimeSpan.FromMinutes(1));

src/EventStore.Core.Tests/TransactionLog/Scavenging/Helpers/ScavengeTestScenario.cs

@@ -95,6 +95,7 @@ public override async Task TestFixtureSetUp() {
 
 			var scavenger = new TFChunkScavenger<TStreamId>(Serilog.Log.Logger, _dbResult.Db, new FakeTFScavengerLog(), tableIndex, ReadIndex,
 				_logFormat.Metastreams,
+				ITransactionFileTrackerFactory.NoOp,
 				unsafeIgnoreHardDeletes: UnsafeIgnoreHardDelete());
 			await scavenger.Scavenge(alwaysKeepScavenged: true, mergeChunks: false);
 		}

src/EventStore.Core.Tests/TransactionLog/when_having_scavenged_tfchunk_with_all_records_removed.cs

@@ -83,7 +83,7 @@ public override async Task TestFixtureSetUp() {
 
 			var scavenger = new TFChunkScavenger<TStreamId>(Serilog.Log.Logger, _db, new FakeTFScavengerLog(), new FakeTableIndex<TStreamId>(),
 				new FakeReadIndex<TLogFormat, TStreamId>(x => EqualityComparer<TStreamId>.Default.Equals(x, streamId), _logFormat.Metastreams),
-				_logFormat.Metastreams);
+				_logFormat.Metastreams, ITransactionFileTrackerFactory.NoOp);
 			await scavenger.Scavenge(alwaysKeepScavenged: true, mergeChunks: false);
 
 			_scavengedChunk = _db.Manager.GetChunk(0);

src/EventStore.Core/ClusterVNode.cs

@@ -1380,6 +1380,7 @@ GossipAdvertiseInfo GetGossipAdvertiseInfo() {
 							tableIndex: tableIndex,
 							readIndex: readIndex,
 							metastreams: logFormat.SystemStreams,
+							tfTrackers: trackers.TransactionFileTrackers,
 							unsafeIgnoreHardDeletes: options.Database.UnsafeIgnoreHardDelete,
 							threads: message.Threads)));
 			}

src/EventStore.Core/Services/UserManagement/SystemAccounts.cs

@@ -18,5 +18,6 @@ public class SystemAccounts {
 		public static readonly string SystemRedactionName = "system-redaction";
 		public static readonly string SystemIndexCommitterName = "system-index-committer";
 		public static readonly string SystemPersistentSubscriptionsName = "system-persistent-subscriptions";
+		public static readonly string SystemScavengeName = "system-scavenge";
 	}
 }

src/EventStore.Core/TransactionLog/Chunks/TFChunkScavenger.cs

@@ -12,6 +12,7 @@
 using EventStore.Core.Index;
 using EventStore.Core.LogAbstraction;
 using EventStore.Core.Services.Storage.ReaderIndex;
+using EventStore.Core.Services.UserManagement;
 using EventStore.Core.TransactionLog.Chunks.TFChunk;
 using EventStore.Core.TransactionLog.LogRecords;
 using EventStore.Core.TransactionLog.Scavenging;
@@ -33,12 +34,15 @@ public class TFChunkScavenger<TStreamId> : TFChunkScavenger {
 		private readonly ITableIndex<TStreamId> _tableIndex;
 		private readonly IReadIndex<TStreamId> _readIndex;
 		private readonly IMetastreamLookup<TStreamId> _metastreams;
+		private readonly ITransactionFileTracker _tfTracker;
 		private readonly long _maxChunkDataSize;
 		private readonly bool _unsafeIgnoreHardDeletes;
 		private readonly int _threads;
 
 		public TFChunkScavenger(ILogger logger, TFChunkDb db, ITFChunkScavengerLog scavengerLog, ITableIndex<TStreamId> tableIndex,
-			IReadIndex<TStreamId> readIndex, IMetastreamLookup<TStreamId> metastreams, long? maxChunkDataSize = null,
+			IReadIndex<TStreamId> readIndex, IMetastreamLookup<TStreamId> metastreams,
+			ITransactionFileTrackerFactory tfTrackers,
+			long? maxChunkDataSize = null,
 			bool unsafeIgnoreHardDeletes = false, int threads = 1) {
 			Ensure.NotNull(logger, nameof(logger));
 			Ensure.NotNull(db, "db");
@@ -62,6 +66,7 @@ public TFChunkScavenger(ILogger logger, TFChunkDb db, ITFChunkScavengerLog scave
 			_tableIndex = tableIndex;
 			_readIndex = readIndex;
 			_metastreams = metastreams;
+			_tfTracker = tfTrackers.GetOrAdd(SystemAccounts.SystemScavengeName);
 			_maxChunkDataSize = maxChunkDataSize ?? db.Config.ChunkSize;
 			_unsafeIgnoreHardDeletes = unsafeIgnoreHardDeletes;
 			_threads = threads;
@@ -158,6 +163,7 @@ private void ScavengeInternal(bool alwaysKeepScavenged, bool mergeChunks, int st
 					maxChunkDataSize: _maxChunkDataSize,
 					scavengerLog: _scavengerLog,
 					throttle: new Throttle(_logger, TimeSpan.Zero, TimeSpan.Zero, 100),
+					tracker: _tfTracker,
 					ct: ct);
 			}
 
@@ -204,7 +210,7 @@ private void ScavengeChunk(bool alwaysKeepScavenged, TFChunk.TFChunk oldChunk,
 			}
 
 			try {
-				TraverseChunkBasic(oldChunk, ct,
+				TraverseChunkBasic(oldChunk, ct, _tfTracker,
 					result => {
 						threadLocalCache.Records.Add(result);
 
@@ -337,6 +343,7 @@ public static void MergePhase(
 			long maxChunkDataSize,
 			ITFChunkScavengerLog scavengerLog,
 			Throttle throttle,
+			ITransactionFileTracker tracker,
 			CancellationToken ct) {
 
 			bool mergedSomething;
@@ -362,6 +369,7 @@ public static void MergePhase(
 								db: db,
 								scavengerLog: scavengerLog,
 								oldChunks: chunksToMerge,
+								tracker: tracker,
 								ct: ct)) {
 
 							mergedSomething = true;
@@ -382,6 +390,7 @@ public static void MergePhase(
 						db: db,
 						scavengerLog: scavengerLog,
 						oldChunks: chunksToMerge,
+						tracker: tracker,
 						ct: ct)) {
 
 						mergedSomething = true;
@@ -398,6 +407,7 @@ private static bool MergeChunks(
 			TFChunkDb db,
 			ITFChunkScavengerLog scavengerLog,
 			IList<TFChunk.TFChunk> oldChunks,
+			ITransactionFileTracker tracker,
 			CancellationToken ct) {
 
 			if (oldChunks.IsEmpty()) throw new ArgumentException("Provided list of chunks to merge is empty.");
@@ -444,7 +454,7 @@ private static bool MergeChunks(
 				var positionMapping = new List<PosMap>();
 				foreach (var oldChunk in oldChunks) {
 					var lastFlushedPage = -1;
-					TraverseChunkBasic(oldChunk, ct,
+					TraverseChunkBasic(oldChunk, ct, tracker,
 						result => {
 
 							positionMapping.Add(WriteRecord(newChunk, result.LogRecord));
@@ -613,7 +623,7 @@ private bool ShouldKeepPrepare(
 				return true;
 			}
 
-			var lastEventNumber = _readIndex.GetStreamLastEventNumber(prepare.EventStreamId, ITransactionFileTracker.NoOp);
+			var lastEventNumber = _readIndex.GetStreamLastEventNumber(prepare.EventStreamId, _tfTracker);
 			if (lastEventNumber == EventNumber.DeletedStream) {
 				// The stream is hard deleted but this is not the tombstone.
 				// When all prepares and commit of transaction belong to single chunk and the stream is deleted,
@@ -671,7 +681,7 @@ private bool ShouldKeepPrepare(
 				return true;
 			}
 
-			var meta = _readIndex.GetStreamMetadata(prepare.EventStreamId, ITransactionFileTracker.NoOp);
+			var meta = _readIndex.GetStreamMetadata(prepare.EventStreamId, _tfTracker);
 			bool canRemove = (meta.MaxCount.HasValue && eventNumber < lastEventNumber - meta.MaxCount.Value + 1)
 			                 || (meta.TruncateBefore.HasValue && eventNumber < meta.TruncateBefore.Value)
 			                 || (meta.MaxAge.HasValue && prepare.TimeStamp < DateTime.UtcNow - meta.MaxAge.Value);
@@ -686,7 +696,7 @@ private bool ShouldKeepPrepare(
 		}
 
 		private bool DiscardBecauseDuplicate(IPrepareLogRecord<TStreamId> prepare, long eventNumber) {
-			var result = _readIndex.ReadEvent(IndexReader.UnspecifiedStreamName, prepare.EventStreamId, eventNumber, ITransactionFileTracker.NoOp);
+			var result = _readIndex.ReadEvent(IndexReader.UnspecifiedStreamName, prepare.EventStreamId, eventNumber, _tfTracker);
 			if (result.Result == ReadEventResult.Success && result.Record.LogPosition != prepare.LogPosition) {
 				// prepare isn't the record we get for an index read at its own stream/version.
 				// therefore it is a duplicate that cannot be read from the index, discard it.
@@ -701,13 +711,13 @@ private bool IsSoftDeletedTempStreamWithinSameChunk(TStreamId eventStreamId, lon
 			TStreamId msh;
 			if (_metastreams.IsMetaStream(eventStreamId)) {
 				var originalStreamId = _metastreams.OriginalStreamOf(eventStreamId);
-				var meta = _readIndex.GetStreamMetadata(originalStreamId, ITransactionFileTracker.NoOp);
+				var meta = _readIndex.GetStreamMetadata(originalStreamId, _tfTracker);
 				if (meta.TruncateBefore != EventNumber.DeletedStream || meta.TempStream != true)
 					return false;
 				sh = originalStreamId;
 				msh = eventStreamId;
 			} else {
-				var meta = _readIndex.GetStreamMetadata(eventStreamId, ITransactionFileTracker.NoOp);
+				var meta = _readIndex.GetStreamMetadata(eventStreamId, _tfTracker);
 				if (meta.TruncateBefore != EventNumber.DeletedStream || meta.TempStream != true)
 					return false;
 				sh = eventStreamId;
@@ -727,14 +737,15 @@ private bool IsSoftDeletedTempStreamWithinSameChunk(TStreamId eventStreamId, lon
 		}
 
 		private static void TraverseChunkBasic(TFChunk.TFChunk chunk, CancellationToken ct,
+			ITransactionFileTracker tracker,
 			Action<CandidateRecord> process) {
-			var result = chunk.TryReadFirst(ITransactionFileTracker.NoOp); //qq
+			var result = chunk.TryReadFirst(tracker);
 			while (result.Success) {
 				process(new CandidateRecord(result.LogRecord, result.RecordLength));
 
 				ct.ThrowIfCancellationRequested();
 
-				result = chunk.TryReadClosestForward(result.NextPosition, ITransactionFileTracker.NoOp); //qq
+				result = chunk.TryReadClosestForward(result.NextPosition, tracker);
 			}
 		}
 

src/EventStore.Core/TransactionLog/Scavenging/DbAccess/OldScavengeChunkMergerBackend.cs

@@ -29,6 +29,7 @@ public void MergeChunks(
 				maxChunkDataSize: _db.Config.ChunkSize,
 				scavengerLog: scavengerLogger,
 				throttle: throttle,
+				tracker: ITransactionFileTracker.NoOp,
 				ct: cancellationToken);
 		}
 	}