<commit_before>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using NBitcoin;
using Stratis.Bitcoin.AsyncWork;
using Stratis.Bitcoin.Base;
using Stratis.Bitcoin.Interfaces;
using Stratis.Bitcoin.Primitives;
using Stratis.Bitcoin.Utilities;
using Stratis.Bitcoin.Utilities.Extensions;

namespace Stratis.Bitcoin.Features.BlockStore
{
    /// <summary>
    /// Saves blocks to the database in batches, removes reorged blocks from the database.
    /// </summary>
    /// <remarks>
    /// <para>
    /// The batch is saved when total serialized size of all blocks in a batch reaches <see cref="BatchThresholdSizeBytes"/>,
    /// or when more than <see cref="BatchMaxSaveIntervalSeconds"/> passed since last batch was saved, or when node is shutting down.
    /// </para>
    /// <para>
    /// When we save new blocks to the database, in case <see cref="IBlockRepository"/> contains blocks that
    /// are no longer a part of our best chain, they are removed from the database.
    /// </para>
    /// <para>
    /// When block store is being initialized we delete blocks that are not on the best chain.
    /// </para>
    /// </remarks>
    public class BlockStoreQueue : IBlockStoreQueue
    {
        /// <summary>Maximum interval between saving batches.</summary>
        /// <remarks>Interval value is a prime number that wasn't used as an interval in any other component. That prevents having CPU consumption spikes.</remarks>
        private const int BatchMaxSaveIntervalSeconds = 37;

        /// <summary>Maximum number of bytes the batch can hold until the downloaded blocks are stored to the disk.</summary>
        internal long BatchThresholdSizeBytes;

        /// <summary>The current batch size in bytes.</summary>
        private long currentBatchSizeBytes;

        /// <summary>The current pending blocks size in bytes.</summary>
        private long blocksQueueSizeBytes;

        /// <summary>The highest stored block in the repository.</summary>
        private ChainedHeader storeTip;

        /// <inheritdoc cref="ILogger"/>
        private readonly ILogger logger;

        private readonly IBlockStoreQueueFlushCondition blockStoreQueueFlushCondition;

        /// <inheritdoc cref="IChainState"/>
        private readonly IChainState chainState;

        /// <inheritdoc cref="StoreSettings"/>
        private readonly StoreSettings storeSettings;

        /// <inheritdoc cref="ChainIndexer"/>
        private readonly ChainIndexer chainIndexer;

        /// <inheritdoc cref="IBlockRepository"/>
        private readonly IBlockRepository blockRepository;

        private readonly IAsyncProvider asyncProvider;

        /// <summary>Queue which contains blocks that should be saved to the database.</summary>
        private readonly IAsyncQueue<ChainedHeaderBlock> blocksQueue;

        /// <summary>Batch of blocks which should be saved in the database.</summary>
        /// <remarks>Write access should be protected by <see cref="blocksCacheLock"/>.</remarks>
        private readonly List<ChainedHeaderBlock> batch;

        /// <summary>Task that runs <see cref="DequeueBlocksContinuouslyAsync"/>.</summary>
        private Task dequeueLoopTask;

        /// <summary>Protects the batch from being modifying while <see cref="GetBlock"/> method is using the batch.</summary>
        private readonly object blocksCacheLock;

        /// <summary>Represents all blocks currently in the queue & pending batch, so that <see cref="GetBlock"/> is able to return a value directly after enqueuing.</summary>
        /// <remarks>Write access should be protected by <see cref="blocksCacheLock"/>.</remarks>
        private readonly Dictionary<uint256, ChainedHeaderBlock> pendingBlocksCache;

        private readonly CancellationTokenSource cancellation;

        /// <inheritdoc/>
        public ChainedHeader BlockStoreCacheTip { get; private set; }

        private Exception saveAsyncLoopException;

        public BlockStoreQueue(
            ChainIndexer chainIndexer,
            IChainState chainState,
            IBlockStoreQueueFlushCondition blockStoreQueueFlushCondition,
            StoreSettings storeSettings,
            IBlockRepository blockRepository,
            ILoggerFactory loggerFactory,
            INodeStats nodeStats,
            IAsyncProvider asyncProvider)
        {
            Guard.NotNull(blockStoreQueueFlushCondition, nameof(blockStoreQueueFlushCondition));
            Guard.NotNull(chainIndexer, nameof(chainIndexer));
            Guard.NotNull(chainState, nameof(chainState));
            Guard.NotNull(storeSettings, nameof(storeSettings));
            Guard.NotNull(loggerFactory, nameof(loggerFactory));
            Guard.NotNull(blockRepository, nameof(blockRepository));
            Guard.NotNull(nodeStats, nameof(nodeStats));

            this.blockStoreQueueFlushCondition = blockStoreQueueFlushCondition;
            this.chainIndexer = chainIndexer;
            this.chainState = chainState;
            this.storeSettings = storeSettings;
            this.blockRepository = blockRepository;
            this.asyncProvider = asyncProvider;
            this.batch = new List<ChainedHeaderBlock>();
            this.blocksCacheLock = new object();
            this.blocksQueue = asyncProvider.CreateAsyncQueue<ChainedHeaderBlock>();
            this.pendingBlocksCache = new Dictionary<uint256, ChainedHeaderBlock>();
            this.logger = loggerFactory.CreateLogger(this.GetType().FullName);
            this.cancellation = new CancellationTokenSource();
            this.saveAsyncLoopException = null;

            this.BatchThresholdSizeBytes = storeSettings.MaxCacheSize * 1024 * 1024;

            nodeStats.RegisterStats(this.AddComponentStats, StatsType.Component);
        }

        /// <summary>
        /// Initializes the <see cref="BlockStoreQueue"/>.
        /// <para>
        /// If <see cref="storeTip"/> is <c>null</c>, the store is out of sync. This can happen when:</para>
        /// <list>
        ///     <item>The node crashed.</item>
        ///     <item>The node was not closed down properly.</item>
        /// </list>
        /// <para>
        /// To recover we walk back the chain until a common block header is found and set the <see cref="BlockStoreQueue"/>'s <see cref="storeTip"/> to that.
        /// </para>
        /// </summary>
        public void Initialize()
        {
            this.blockRepository.Initialize();

            if (this.storeSettings.ReIndex)
            {
                this.blockRepository.SetTxIndex(this.storeSettings.TxIndex);
                this.blockRepository.ReIndex();
            }

            ChainedHeader initializationTip = this.chainIndexer.GetHeader(this.blockRepository.TipHashAndHeight.Hash);
            this.SetStoreTip(initializationTip);

            if (this.storeTip == null)
                this.RecoverStoreTip();

            this.logger.LogDebug("Initialized block store tip at '{0}'.", this.storeTip);

            if (this.storeSettings.TxIndex != this.blockRepository.TxIndex)
            {
                if (this.storeTip != this.chainIndexer.Genesis)
                {
                    this.logger.LogTrace("(-)[REBUILD_REQUIRED]");
                    throw new BlockStoreException("You need to rebuild the block store database using -reindex to change -txindex");
                }

                // We only reach here in the case where we are syncing with a database with no blocks.
                // Always set the TxIndex here.
                this.blockRepository.SetTxIndex(this.storeSettings.TxIndex);
            }

            // Throw if block store was initialized after the consensus.
            // This is needed in order to rewind consensus in case it is initialized ahead of the block store.
            if (this.chainState.ConsensusTip != null)
            {
                this.logger.LogCritical("Block store initialized after the consensus!");
                this.logger.LogTrace("(-)[INITIALIZATION_ERROR]");
                throw new BlockStoreException("Block store initialized after consensus!");
            }

            this.BlockStoreCacheTip = initializationTip;

            // Start dequeuing.
            this.currentBatchSizeBytes = 0;
            this.dequeueLoopTask = this.DequeueBlocksContinuouslyAsync();
        }

        /// <inheritdoc/>
        public Transaction GetTransactionById(uint256 trxid)
        {
            // Only look for transactions if they're indexed.
            if (!this.storeSettings.TxIndex)
            {
                this.logger.LogTrace("(-)[TX_INDEX_DISABLED]:null");<commit_msg>Add block store task to async provider (#3670)<commit_after>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using NBitcoin;
using Stratis.Bitcoin.AsyncWork;
using Stratis.Bitcoin.Base;
using Stratis.Bitcoin.Interfaces;
using Stratis.Bitcoin.Primitives;
using Stratis.Bitcoin.Utilities;
using Stratis.Bitcoin.Utilities.Extensions;

namespace Stratis.Bitcoin.Features.BlockStore
{
    /// <summary>
    /// Saves blocks to the database in batches, removes reorged blocks from the database.
    /// </summary>
    /// <remarks>
    /// <para>
    /// The batch is saved when total serialized size of all blocks in a batch reaches <see cref="BatchThresholdSizeBytes"/>,
    /// or when more than <see cref="BatchMaxSaveIntervalSeconds"/> passed since last batch was saved, or when node is shutting down.
    /// </para>
    /// <para>
    /// When we save new blocks to the database, in case <see cref="IBlockRepository"/> contains blocks that
    /// are no longer a part of our best chain, they are removed from the database.
    /// </para>
    /// <para>
    /// When block store is being initialized we delete blocks that are not on the best chain.
    /// </para>
    /// </remarks>
    public class BlockStoreQueue : IBlockStoreQueue
    {
        /// <summary>Maximum interval between saving batches.</summary>
        /// <remarks>Interval value is a prime number that wasn't used as an interval in any other component. That prevents having CPU consumption spikes.</remarks>
        private const int BatchMaxSaveIntervalSeconds = 37;

        /// <summary>Maximum number of bytes the batch can hold until the downloaded blocks are stored to the disk.</summary>
        internal long BatchThresholdSizeBytes;

        /// <summary>The current batch size in bytes.</summary>
        private long currentBatchSizeBytes;

        /// <summary>The current pending blocks size in bytes.</summary>
        private long blocksQueueSizeBytes;

        /// <summary>The highest stored block in the repository.</summary>
        private ChainedHeader storeTip;

        /// <inheritdoc cref="ILogger"/>
        private readonly ILogger logger;

        private readonly IBlockStoreQueueFlushCondition blockStoreQueueFlushCondition;

        /// <inheritdoc cref="IChainState"/>
        private readonly IChainState chainState;

        /// <inheritdoc cref="StoreSettings"/>
        private readonly StoreSettings storeSettings;

        /// <inheritdoc cref="ChainIndexer"/>
        private readonly ChainIndexer chainIndexer;

        /// <inheritdoc cref="IBlockRepository"/>
        private readonly IBlockRepository blockRepository;

        private readonly IAsyncProvider asyncProvider;

        /// <summary>Queue which contains blocks that should be saved to the database.</summary>
        private readonly IAsyncQueue<ChainedHeaderBlock> blocksQueue;

        /// <summary>Batch of blocks which should be saved in the database.</summary>
        /// <remarks>Write access should be protected by <see cref="blocksCacheLock"/>.</remarks>
        private readonly List<ChainedHeaderBlock> batch;

        /// <summary>Task that runs <see cref="DequeueBlocksContinuouslyAsync"/>.</summary>
        private Task dequeueLoopTask;

        /// <summary>Protects the batch from being modifying while <see cref="GetBlock"/> method is using the batch.</summary>
        private readonly object blocksCacheLock;

        /// <summary>Represents all blocks currently in the queue & pending batch, so that <see cref="GetBlock"/> is able to return a value directly after enqueuing.</summary>
        /// <remarks>Write access should be protected by <see cref="blocksCacheLock"/>.</remarks>
        private readonly Dictionary<uint256, ChainedHeaderBlock> pendingBlocksCache;

        private readonly CancellationTokenSource cancellation;

        /// <inheritdoc/>
        public ChainedHeader BlockStoreCacheTip { get; private set; }

        private Exception saveAsyncLoopException;

        public BlockStoreQueue(
            ChainIndexer chainIndexer,
            IChainState chainState,
            IBlockStoreQueueFlushCondition blockStoreQueueFlushCondition,
            StoreSettings storeSettings,
            IBlockRepository blockRepository,
            ILoggerFactory loggerFactory,
            INodeStats nodeStats,
            IAsyncProvider asyncProvider)
        {
            Guard.NotNull(blockStoreQueueFlushCondition, nameof(blockStoreQueueFlushCondition));
            Guard.NotNull(chainIndexer, nameof(chainIndexer));
            Guard.NotNull(chainState, nameof(chainState));
            Guard.NotNull(storeSettings, nameof(storeSettings));
            Guard.NotNull(loggerFactory, nameof(loggerFactory));
            Guard.NotNull(blockRepository, nameof(blockRepository));
            Guard.NotNull(nodeStats, nameof(nodeStats));

            this.blockStoreQueueFlushCondition = blockStoreQueueFlushCondition;
            this.chainIndexer = chainIndexer;
            this.chainState = chainState;
            this.storeSettings = storeSettings;
            this.blockRepository = blockRepository;
            this.asyncProvider = asyncProvider;
            this.batch = new List<ChainedHeaderBlock>();
            this.blocksCacheLock = new object();
            this.blocksQueue = asyncProvider.CreateAsyncQueue<ChainedHeaderBlock>();
            this.pendingBlocksCache = new Dictionary<uint256, ChainedHeaderBlock>();
            this.logger = loggerFactory.CreateLogger(this.GetType().FullName);
            this.cancellation = new CancellationTokenSource();
            this.saveAsyncLoopException = null;

            this.BatchThresholdSizeBytes = storeSettings.MaxCacheSize * 1024 * 1024;

            nodeStats.RegisterStats(this.AddComponentStats, StatsType.Component);
        }

        /// <summary>
        /// Initializes the <see cref="BlockStoreQueue"/>.
        /// <para>
        /// If <see cref="storeTip"/> is <c>null</c>, the store is out of sync. This can happen when:</para>
        /// <list>
        ///     <item>The node crashed.</item>
        ///     <item>The node was not closed down properly.</item>
        /// </list>
        /// <para>
        /// To recover we walk back the chain until a common block header is found and set the <see cref="BlockStoreQueue"/>'s <see cref="storeTip"/> to that.
        /// </para>
        /// </summary>
        public void Initialize()
        {
            this.blockRepository.Initialize();

            if (this.storeSettings.ReIndex)
            {
                this.blockRepository.SetTxIndex(this.storeSettings.TxIndex);
                this.blockRepository.ReIndex();
            }

            ChainedHeader initializationTip = this.chainIndexer.GetHeader(this.blockRepository.TipHashAndHeight.Hash);
            this.SetStoreTip(initializationTip);

            if (this.storeTip == null)
                this.RecoverStoreTip();

            this.logger.LogDebug("Initialized block store tip at '{0}'.", this.storeTip);

            if (this.storeSettings.TxIndex != this.blockRepository.TxIndex)
            {
                if (this.storeTip != this.chainIndexer.Genesis)
                {
                    this.logger.LogTrace("(-)[REBUILD_REQUIRED]");
                    throw new BlockStoreException("You need to rebuild the block store database using -reindex to change -txindex");
                }

                // We only reach here in the case where we are syncing with a database with no blocks.
                // Always set the TxIndex here.
                this.blockRepository.SetTxIndex(this.storeSettings.TxIndex);
            }

            // Throw if block store was initialized after the consensus.
            // This is needed in order to rewind consensus in case it is initialized ahead of the block store.
            if (this.chainState.ConsensusTip != null)
            {
                this.logger.LogCritical("Block store initialized after the consensus!");
                this.logger.LogTrace("(-)[INITIALIZATION_ERROR]");
                throw new BlockStoreException("Block store initialized after consensus!");
            }

            this.BlockStoreCacheTip = initializationTip;

            // Start dequeuing.
            this.currentBatchSizeBytes = 0;
            this.dequeueLoopTask = this.DequeueBlocksContinuouslyAsync();

            this.asyncProvider.RegisterTask($"{nameof(BlockStoreQueue)}.{nameof(this.dequeueLoopTask)}", this.dequeueLoopTask);
        }

        /// <inheritdoc/>
        public Transaction GetTransactionById(uint256 trxid)
        {
            // Only look for transactions if they're indexed.
            if (!this.storeSettings.TxIndex)
            {
                this.logger.LogTrace("(-)[TX_INDEX_DISABLED]:null");
{'fraction_non_alphanumeric': 0.09388054646042678, 'fraction_numerical': 0.002596816077678672, 'mean_word_length': 2.4599609375, 'pattern_counts': {'":': 0, '<': 153, '<?xml version=': 0, '>': 153, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '254448', 'n_tokens_mistral': 4340, 'n_tokens_neox': 4200, 'n_words': 1272}