fix index_in_epoch value for batch processed epoch changes.

Author: PawelPawelec-RDX

CHANGELOG.md

@@ -1,6 +1,9 @@
 ## 1.11.0
 Release built: _not released yet_
 
+### Database changes
+- Fixed an invalid `index_in_epoch` value for transactions processed in a batch when they belonged to different epochs.
+
 ## 1.10.1
 Release built: 6.03.2025
 

src/RadixDlt.NetworkGateway.DataAggregator/Services/ITopOfLedgerProvider.cs

@@ -72,4 +72,6 @@ public interface ITopOfLedgerProvider
     Task<TransactionSummary> GetTopOfLedger(CancellationToken token);
 
     Task<long> GetLastCommittedStateVersion(CancellationToken token);
+
+    Task<long> GetLastCommittedEpoch(CancellationToken token);
 }

src/RadixDlt.NetworkGateway.DataAggregator/Services/LedgerTransactionsProcessor.cs

@@ -92,7 +92,6 @@ public sealed class LedgerTransactionsProcessor : ILedgerTransactionsProcessor
     private readonly IEnumerable<ILedgerConfirmationServiceObserver> _observers;
     private readonly ITopOfLedgerProvider _topOfLedgerProvider;
     private readonly IFetchedTransactionStore _fetchedTransactionStore;
-    private readonly INetworkConfigurationProvider _networkConfigurationProvider;
     private readonly IClock _clock;
 
     private LedgerConfirmationOptions Config { get; set; }
@@ -105,7 +104,6 @@ public LedgerTransactionsProcessor(
         IEnumerable<ILedgerConfirmationServiceObserver> observers,
         IFetchedTransactionStore fetchedTransactionStore,
         ITopOfLedgerProvider topOfLedgerProvider,
-        INetworkConfigurationProvider networkConfigurationProvider,
         IClock clock)
     {
         _logger = logger;
@@ -116,19 +114,17 @@ public LedgerTransactionsProcessor(
         _fetchedTransactionStore = fetchedTransactionStore;
         _topOfLedgerProvider = topOfLedgerProvider;
         _clock = clock;
-        _networkConfigurationProvider = networkConfigurationProvider;
         Config = _ledgerConfirmationOptionsMonitor.CurrentValue;
     }
 
     public async Task ProcessTransactions(CancellationToken token)
     {
-        var networkConfiguration = _networkConfigurationProvider.GetNetworkConfiguration();
         var lastCommittedTransactionSummary = await _topOfLedgerProvider.GetTopOfLedger(token);
         await _observers.ForEachAsync(x => x.PreHandleLedgerExtension(_clock.UtcNow));
 
         Config = _ledgerConfirmationOptionsMonitor.CurrentValue;
 
-        var transactions = ConstructLedgerExtension(lastCommittedTransactionSummary);
+        var transactions = ConstructLedgerExtension(lastCommittedTransactionSummary.StateVersion);
 
         if (transactions.Count == 0)
         {
@@ -146,9 +142,9 @@ public async Task ProcessTransactions(CancellationToken token)
         await DelayBetweenIngestionBatchesIfRequested(commitReport);
     }
 
-    private List<CoreModel.CommittedTransaction> ConstructLedgerExtension(TransactionSummary topOfLedger)
+    private List<CoreModel.CommittedTransaction> ConstructLedgerExtension(long topOfLedgerStateVersion)
     {
-        var startStateVersion = topOfLedger.StateVersion + 1;
+        var startStateVersion = topOfLedgerStateVersion + 1;
         var transactions = _fetchedTransactionStore.GetTransactionBatch(startStateVersion, (int)Config.MaxCommitBatchSize, (int)Config.MinCommitBatchSize);
         return transactions;
     }

src/RadixDlt.NetworkGateway.DataAggregator/Workers/NodeWorkers/NodeTransactionFetchWorker.cs

@@ -278,8 +278,7 @@ private async Task<FetchedTransactions> FetchTransactions(long fromStateVersion,
 
     private async Task<(long StateVersionInclusiveLowerBound, long StateVersionInclusiveUpperBound)?> GetTransactionsToFetch(string nodeName, CancellationToken cancellationToken)
     {
-        var lastCommittedTransactionSummary = await _topOfLedgerProvider.GetTopOfLedger(cancellationToken);
-        var processTransactionFromStateVersion = lastCommittedTransactionSummary.StateVersion;
+        var processTransactionFromStateVersion = await _topOfLedgerProvider.GetLastCommittedStateVersion(cancellationToken);
         var maxUpperLimit = processTransactionFromStateVersion + _ledgerConfirmationOptionsMonitor.CurrentValue.MaxTransactionPipelineSizePerNode + 1;
 
         var shouldFetchNewTransactions = _fetchedTransactionStore.ShouldFetchNewTransactions(nodeName, processTransactionFromStateVersion);

src/RadixDlt.NetworkGateway.PostgresIntegration/LedgerExtension/PostgresLedgerExtenderService.cs

@@ -131,9 +131,9 @@ public async Task<CommitTransactionsReport> CommitTransactions(ConsistentLedgerE
 
         try
         {
-            var topOfLedgerSummary = await _topOfLedgerProvider.GetTopOfLedger(token);
+            var lastCommittedStateVersion = await _topOfLedgerProvider.GetLastCommittedStateVersion(token);
 
-            TransactionConsistencyValidator.AssertLatestTransactionConsistent(ledgerExtension.LatestTransactionSummary.StateVersion, topOfLedgerSummary.StateVersion);
+            TransactionConsistencyValidator.AssertLatestTransactionConsistent(ledgerExtension.LatestTransactionSummary.StateVersion, lastCommittedStateVersion);
 
             var extendLedgerReport = await ProcessTransactions(dbContext, ledgerExtension, token);
 

src/RadixDlt.NetworkGateway.PostgresIntegration/LedgerExtension/Processors/LedgerTransactionProcessor.cs

@@ -76,6 +76,7 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
+using System.Runtime.InteropServices.JavaScript;
 using System.Threading.Tasks;
 using CoreModel = RadixDlt.CoreApiSdk.Model;
 
@@ -89,11 +90,11 @@ private record TransactionData(
         CoreModel.CommittedTransaction RawCommittedTransaction
     )
     {
-        public long? NewEpochIndex { get; set; }
+        public long? RoundUpdateEpochIndex { get; set; }
 
         public long? NewRoundIndex { get; set; }
 
-        public DateTime? RoundTimestampUpdate { get; set; }
+        public DateTime? NewRoundTimestamp { get; set; }
     }
 
     private readonly ProcessorContext _context;
@@ -134,15 +135,13 @@ public void VisitTransaction(CoreModel.CommittedTransaction committedTransaction
 
         if (committedTransaction.LedgerTransaction is CoreModel.RoundUpdateLedgerTransaction roundUpdateTransaction)
         {
-            var newEpochIndex = _lastProcessedTransactionSummary.Epoch != roundUpdateTransaction.RoundUpdateTransaction.Epoch ? roundUpdateTransaction.RoundUpdateTransaction.Epoch : (long?)null;
-
             _transactionData.Update(
                 stateVersion,
                 existing =>
                 {
-                    existing.NewEpochIndex = newEpochIndex;
+                    existing.RoundUpdateEpochIndex = roundUpdateTransaction.RoundUpdateTransaction.Epoch;
                     existing.NewRoundIndex = roundUpdateTransaction.RoundUpdateTransaction.RoundInEpoch;
-                    existing.RoundTimestampUpdate = DateTimeOffset.FromUnixTimeMilliseconds(roundUpdateTransaction.RoundUpdateTransaction.ProposerTimestamp.UnixTimestampMs).UtcDateTime;
+                    existing.NewRoundTimestamp = DateTimeOffset.FromUnixTimeMilliseconds(roundUpdateTransaction.RoundUpdateTransaction.ProposerTimestamp.UnixTimestampMs).UtcDateTime;
                 });
         }
     }
@@ -155,7 +154,7 @@ public void VisitUpsert(CoreModel.IUpsertedSubstate substate, ReferencedEntity r
         {
             _transactionData.Update(
                 stateVersion,
-                existing => { existing.RoundTimestampUpdate = DateTimeOffset.FromUnixTimeMilliseconds(currentTime.Value.ProposerTimestamp.UnixTimestampMs).UtcDateTime; });
+                existing => { existing.NewRoundTimestamp = DateTimeOffset.FromUnixTimeMilliseconds(currentTime.Value.ProposerTimestamp.UnixTimestampMs).UtcDateTime; });
         }
     }
 
@@ -171,17 +170,18 @@ public void ProcessChanges()
             var stateVersion = data.StateVersion;
             var committedTransaction = data.RawCommittedTransaction;
 
-            var roundTimestamp = data.RoundTimestampUpdate ?? _lastProcessedTransactionSummary.RoundTimestamp;
-
+            var roundInEpoch = data.NewRoundIndex ?? _lastProcessedTransactionSummary.RoundInEpoch;
+            var roundTimestamp = data.NewRoundTimestamp ?? _lastProcessedTransactionSummary.RoundTimestamp;
             var normalizedRoundTimestamp =
                 roundTimestamp < _lastProcessedTransactionSummary.NormalizedRoundTimestamp ? _lastProcessedTransactionSummary.NormalizedRoundTimestamp
                 : roundTimestamp > data.CreatedAt ? data.CreatedAt
                 : roundTimestamp;
 
-            var epoch = data.NewEpochIndex ?? _lastProcessedTransactionSummary.Epoch;
-            var roundInEpoch = data.NewRoundIndex ?? _lastProcessedTransactionSummary.RoundInEpoch;
-            var indexInEpoch = data.NewEpochIndex.HasValue ? 0 : _lastProcessedTransactionSummary.IndexInEpoch + 1;
-            var indexInRound = data.NewEpochIndex.HasValue ? 0 : _lastProcessedTransactionSummary.IndexInRound + 1;
+            var epoch = data.RoundUpdateEpochIndex ?? _lastProcessedTransactionSummary.Epoch;
+            bool isNewEpoch = epoch != _lastProcessedTransactionSummary.Epoch;
+
+            var indexInEpoch = isNewEpoch ? 0 : _lastProcessedTransactionSummary.IndexInEpoch + 1;
+            var indexInRound = data.NewRoundIndex.HasValue ? 0 : _lastProcessedTransactionSummary.IndexInRound + 1;
 
             LedgerTransaction ledgerTransaction = committedTransaction.LedgerTransaction switch
             {
@@ -277,7 +277,7 @@ public void ProcessChanges()
             }
 
             var isUserTransaction = committedTransaction.LedgerTransaction is CoreModel.UserLedgerTransaction or CoreModel.UserLedgerTransactionV2;
-            var isUserTransactionOrEpochChange = isUserTransaction || data.NewEpochIndex.HasValue;
+            var isUserTransactionOrEpochChange = isUserTransaction || isNewEpoch;
 
             if (_context.StorageOptions.StoreReceiptStateUpdates == LedgerTransactionStorageOption.StoreForAllTransactions ||
                 (_context.StorageOptions.StoreReceiptStateUpdates == LedgerTransactionStorageOption.StoreOnlyForUserTransactions && isUserTransaction) ||

src/RadixDlt.NetworkGateway.PostgresIntegration/Services/PendingTransactions/PendingTransactionResubmissionService.cs

@@ -302,8 +302,7 @@ private async Task<ContextualSubmissionResult> Resubmit(PendingTransactionWithCh
 
     private async Task<long> GetCurrentEpoch(CancellationToken cancellationToken)
     {
-        var topOfLedger = await _topOfLedgerProvider.GetTopOfLedger(cancellationToken);
-        var signedEpoch = topOfLedger.Epoch;
-        return (signedEpoch >= 0) ? signedEpoch : throw new InvalidStateException($"Epoch was negative: {signedEpoch}");
+        var lastCommitedEpoch = await _topOfLedgerProvider.GetLastCommittedEpoch(cancellationToken);
+        return (lastCommitedEpoch >= 0) ? lastCommitedEpoch : throw new InvalidStateException($"Epoch was negative: {lastCommitedEpoch}");
     }
 }

src/RadixDlt.NetworkGateway.PostgresIntegration/Services/TopOfLedgerProvider.cs

@@ -87,12 +87,12 @@ public TopOfLedgerProvider(Abstractions.IClock clock, INetworkConfigurationProvi
 
     public async Task<long> GetLastCommittedStateVersion(CancellationToken token)
     {
-        var dbContext = await _dbContextFactory.CreateDbContextAsync(token);
+        return (await GetTopOfLedger(token)).StateVersion;
+    }
 
-        return await dbContext.GetTopLedgerTransaction()
-            .AsNoTracking()
-            .Select(t => t.StateVersion)
-            .FirstOrDefaultAsync(token); // Defaults to 0, which is perfect
+    public async Task<long> GetLastCommittedEpoch(CancellationToken token)
+    {
+        return (await GetTopOfLedger(token)).Epoch;
     }
 
     public async Task<TransactionSummary> GetTopOfLedger(CancellationToken token)