blockbook.go

package main

import (
	"context"
	"flag"
	"log"
	"math/rand"
	"net/http"
	_ "net/http/pprof"
	"os"
	"os/signal"
	"runtime/debug"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/golang/glog"
	"github.com/juju/errors"
	"github.com/trezor/blockbook/api"
	"github.com/trezor/blockbook/bchain"
	"github.com/trezor/blockbook/bchain/coins"
	"github.com/trezor/blockbook/common"
	"github.com/trezor/blockbook/db"
	"github.com/trezor/blockbook/fiat"
	"github.com/trezor/blockbook/fourbyte"
	"github.com/trezor/blockbook/server"
)

// default debounce for too-close requests for resync
const defaultResyncIndexDebounceMs = 1009

// debounce too close requests for resync mempool (ZeroMQ sends message for each tx, when new block there are many transactions)
const debounceResyncMempoolMs = 1009

// store internal state about once every minute
const storeInternalStatePeriodMs = 59699

// the first internal data healing pass runs soon after the initial sync, while the backend
// cache is still warm from the fetches that failed during the previous run, then hourly
const healInternalDataStartupDelay = time.Minute
const healInternalDataPeriod = time.Hour

// exit codes from the main function
const exitCodeOK = 0
const exitCodeFatal = 255

var (
	configFile = flag.String("blockchaincfg", "", "path to blockchain RPC service configuration json file")

	dbPath         = flag.String("datadir", "./data", "path to database directory")
	dbCache        = flag.Int("dbcache", 1<<29, "size of the rocksdb cache")
	dbMaxOpenFiles = flag.Int("dbmaxopenfiles", 1<<14, "max open files by rocksdb")

	blockFrom      = flag.Int("blockheight", -1, "height of the starting block")
	blockUntil     = flag.Int("blockuntil", -1, "height of the final block")
	rollbackHeight = flag.Int("rollback", -1, "rollback to the given height and quit")

	synchronize = flag.Bool("sync", false, "synchronizes until tip, if together with zeromq, keeps index synchronized")
	repair      = flag.Bool("repair", false, "repair physical RocksDB corruption and exit; does NOT recover an interrupted bulk import (use -fixutxo for UTXO index drift)")
	forceRepair = flag.Bool("forcerepair", false, "with -repair, force a database left in inconsistent state (interrupted initial/bulk import) into open state; the index may be incomplete and a full resync is strongly advised")
	fixUtxo     = flag.Bool("fixutxo", false, "check and fix utxo db and exit")
	prof        = flag.String("prof", "", "http server binding [address]:port of the interface to profiling data /debug/pprof/ (default no profiling)")

	syncChunk   = flag.Int("chunk", 100, "block chunk size for processing in bulk mode")
	syncWorkers = flag.Int("workers", 8, "number of workers to process blocks in bulk mode")
	dryRun      = flag.Bool("dryrun", false, "do not index blocks, only download")

	debugMode = flag.Bool("debug", false, "debug mode, return more verbose errors, reload templates on each request")

	internalBinding = flag.String("internal", "", "internal http server binding [address]:port, (default no internal server)")

	publicBinding = flag.String("public", "", "public http server binding [address]:port[/path] (default no public server)")

	certFiles = flag.String("certfile", "", "to enable SSL specify path to certificate files without extension, expecting <certfile>.crt and <certfile>.key (default no SSL)")

	explorerURL = flag.String("explorer", "", "address of blockchain explorer")

	noTxCache = flag.Bool("notxcache", false, "disable tx cache")

	enableSubNewTx = flag.Bool("enablesubnewtx", false, "enable support for subscribing to all new transactions")

	computeColumnStats  = flag.Bool("computedbstats", false, "compute column stats and exit")
	computeFeeStatsFlag = flag.Bool("computefeestats", false, "compute fee stats for blocks in blockheight-blockuntil range and exit")
	dbStatsPeriodHours  = flag.Int("dbstatsperiod", 24, "period of db stats collection in hours, 0 disables stats collection")

	// resync index at least each resyncIndexPeriodMs (could be more often if invoked by message from ZeroMQ)
	resyncIndexPeriodMs = flag.Int("resyncindexperiod", 935093, "resync index period in milliseconds")

	// debounce for push-triggered index resync requests
	resyncIndexDebounceMs = flag.Int("resyncindexdebounce", defaultResyncIndexDebounceMs, "debounce for push-triggered index resync requests in milliseconds")

	// resync mempool at least each resyncMempoolPeriodMs (could be more often if invoked by message from ZeroMQ)
	resyncMempoolPeriodMs = flag.Int("resyncmempoolperiod", 60017, "resync mempool period in milliseconds")

	extendedIndex = flag.Bool("extendedindex", false, "if true, create index of input txids and spending transactions")
)

var (
	// Buffer 1 + non-blocking sends in pushSynchronizationHandler decouple the
	// WebSocket/ZMQ source goroutines from sync progress. If sync is stuck inside
	// TickAndDebounce's f(), additional notifications are dropped here rather
	// than blocking the source; TickAndDebounce's tick-period backstop still fires.
	chanSyncIndex                 = make(chan struct{}, 1)
	chanSyncMempool               = make(chan struct{}, 1)
	chanStoreInternalState        = make(chan struct{})
	chanSyncIndexDone             = make(chan struct{})
	chanSyncMempoolDone           = make(chan struct{})
	chanStoreInternalStateDone    = make(chan struct{})
	chanHealInternalDataDone      = make(chan struct{})
	chain                         bchain.BlockChain
	mempool                       bchain.Mempool
	index                         *db.RocksDB
	txCache                       *db.TxCache
	metrics                       *common.Metrics
	syncWorker                    *db.SyncWorker
	internalState                 *common.InternalState
	fiatRates                     *fiat.FiatRates
	callbacksOnNewBlock           []bchain.OnNewBlockFunc
	callbacksOnNewTx              []bchain.OnNewTxFunc
	callbacksOnNewFiatRatesTicker []fiat.OnNewFiatRatesTicker
	chanOsSignal                  chan os.Signal
)

func init() {
	glog.MaxSize = 1024 * 1024 * 8
	glog.CopyStandardLogTo("INFO")
}

func main() {
	defer func() {
		if e := recover(); e != nil {
			glog.Error("main recovered from panic: ", e)
			debug.PrintStack()
			os.Exit(-1)
		}
	}()
	os.Exit(mainWithExitCode())
}

// allow deferred functions to run even in case of fatal error
func mainWithExitCode() int {
	flag.Parse()

	defer glog.Flush()

	rand.Seed(time.Now().UTC().UnixNano())

	chanOsSignal = make(chan os.Signal, 1)
	shutdownSigCh := make(chan os.Signal, 1)
	signalCh := make(chan os.Signal, 1)
	// Use a single signal listener and fan out shutdown signals to avoid races
	// where long-running workers consume the OS signal before main shutdown runs.
	signal.Notify(signalCh, syscall.SIGHUP, syscall.SIGINT, syscall.SIGQUIT, syscall.SIGTERM)
	var shutdownOnce sync.Once
	go func() {
		sig := <-signalCh
		shutdownOnce.Do(func() {
			// Flip global shutdown state and close chanOsSignal to broadcast shutdown.
			// Closing the channel unblocks select loops that only receive from it.
			common.SetInShutdown()
			close(chanOsSignal)
			// Ensure waitForSignalAndShutdown can proceed even if the OS signal
			// was already consumed by another goroutine in previous versions.
			shutdownSigCh <- sig
		})
	}()

	glog.Infof("Blockbook: %+v, debug mode %v", common.GetVersionInfo(), *debugMode)

	if *prof != "" {
		go func() {
			log.Println(http.ListenAndServe(*prof, nil))
		}()
	}

	if *forceRepair && !*repair {
		glog.Error("-forcerepair has no effect without -repair; re-run with both flags to force an inconsistent database into open state")
		return exitCodeFatal
	}

	if *repair {
		if err := db.RepairRocksDB(*dbPath, *forceRepair); err != nil {
			glog.Errorf("RepairRocksDB %s: %v", *dbPath, err)
			return exitCodeFatal
		}
		return exitCodeOK
	}

	config, err := common.GetConfig(*configFile)
	if err != nil {
		glog.Error("config: ", err)
		return exitCodeFatal
	}

	metrics, err = common.GetMetrics(config.CoinName)
	if err != nil {
		glog.Error("metrics: ", err)
		return exitCodeFatal
	}

	if chain, mempool, err = getBlockChainWithRetry(config.CoinName, *configFile, pushSynchronizationHandler, metrics, 120); err != nil {
		glog.Error("rpc: ", err)
		return exitCodeFatal
	}

	// Chains that expose a configured average block time (currently EVM coins via
	// EthereumRPC.AverageBlockTimeDuration) publish it as a static gauge so
	// alerts can normalize the tip-age metric across coins with different cadences.
	if provider, ok := chain.(interface {
		AverageBlockTimeDuration() (time.Duration, error)
	}); ok {
		if d, err := provider.AverageBlockTimeDuration(); err == nil && d > 0 {
			metrics.AverageBlockTimeSeconds.Set(d.Seconds())
		}
	}

	index, err = db.NewRocksDB(*dbPath, *dbCache, *dbMaxOpenFiles, chain.GetChainParser(), metrics, *extendedIndex)
	if err != nil {
		glog.Error("rocksDB: ", err)
		return exitCodeFatal
	}
	defer func() {
		glog.Info("shutdown: rocksdb close start")
		if err := index.Close(); err != nil {
			glog.Error("shutdown: rocksdb close error: ", err)
		}
		glog.Info("shutdown: rocksdb close finished")
	}()

	internalState, err = newInternalState(config, index, *enableSubNewTx)
	if err != nil {
		glog.Error("internalState: ", err)
		return exitCodeFatal
	}

	// fix possible inconsistencies in the UTXO index
	if *fixUtxo || !internalState.UtxoChecked {
		err = index.FixUtxos(chanOsSignal)
		if err != nil {
			glog.Error("fixUtxos: ", err)
			return exitCodeFatal
		}
		internalState.UtxoChecked = true
	}

	// sort addressContracts if necessary
	if !internalState.SortedAddressContracts {
		err = index.SortAddressContracts(chanOsSignal)
		if err != nil {
			glog.Error("sortAddressContracts: ", err)
			return exitCodeFatal
		}
		internalState.SortedAddressContracts = true
	}

	index.SetInternalState(internalState)
	if *fixUtxo {
		err = index.StoreInternalState(internalState)
		if err != nil {
			glog.Error("StoreInternalState: ", err)
			return exitCodeFatal
		}
		return exitCodeOK
	}

	if internalState.DbState != common.DbStateClosed {
		if internalState.DbState == common.DbStateInconsistent {
			glog.Error("internalState: database is in inconsistent state and cannot be used")
			return exitCodeFatal
		}
		glog.Warning("internalState: database was left in open state, possibly previous ungraceful shutdown")
	}

	if *computeFeeStatsFlag {
		internalState.DbState = common.DbStateOpen
		err = computeFeeStats(chanOsSignal, *blockFrom, *blockUntil, index, chain, txCache, internalState, metrics)
		if err != nil && err != db.ErrOperationInterrupted {
			glog.Error("computeFeeStats: ", err)
			return exitCodeFatal
		}
		return exitCodeOK
	}

	if *computeColumnStats {
		internalState.DbState = common.DbStateOpen
		err = index.ComputeInternalStateColumnStats(chanOsSignal)
		if err != nil {
			glog.Error("internalState: ", err)
			return exitCodeFatal
		}
		glog.Info("DB size on disk: ", index.DatabaseSizeOnDisk(), ", DB size as computed: ", internalState.DBSizeTotal())
		return exitCodeOK
	}

	// Per-chain missing-block retry override, if any. Coin RPCs that opt in
	// expose MissingBlockRetryOverride(); chains without the method keep defaults.
	var syncCfg *db.SyncWorkerConfig
	if provider, ok := chain.(interface {
		MissingBlockRetryOverride() *bchain.MissingBlockRetry
	}); ok {
		if override := provider.MissingBlockRetryOverride(); override != nil {
			syncCfg = &db.SyncWorkerConfig{
				MissingBlockRetry: db.ApplyMissingBlockRetryOverride(override),
			}
			glog.Infof("sync: missingBlockRetry override applied: retryDelay=%s recheckThreshold=%d tipRecheckThreshold=%d maxStall=%s",
				syncCfg.MissingBlockRetry.RetryDelay,
				syncCfg.MissingBlockRetry.RecheckThreshold,
				syncCfg.MissingBlockRetry.TipRecheckThreshold,
				syncCfg.MissingBlockRetry.MaxStallDuration)
		}
	}
	syncWorker, err = db.NewSyncWorkerWithConfig(index, chain, *syncWorkers, *syncChunk, *blockFrom, *dryRun, chanOsSignal, metrics, internalState, syncCfg)
	if err != nil {
		glog.Errorf("NewSyncWorker %v", err)
		return exitCodeFatal
	}

	// set the DbState to open at this moment, after all important workers are initialized
	internalState.DbState = common.DbStateOpen
	err = index.StoreInternalState(internalState)
	if err != nil {
		glog.Error("internalState: ", err)
		return exitCodeFatal
	}

	if *rollbackHeight >= 0 {
		err = performRollback()
		if err != nil {
			return exitCodeFatal
		}
		return exitCodeOK
	}

	if txCache, err = db.NewTxCache(index, chain, metrics, internalState, !*noTxCache); err != nil {
		glog.Error("txCache ", err)
		return exitCodeFatal
	}

	if fiatRates, err = fiat.NewFiatRates(index, config, metrics, onNewFiatRatesTicker); err != nil {
		glog.Error("fiatRates ", err)
		return exitCodeFatal
	}

	// report BlockbookAppInfo metric, only log possible error
	if err = blockbookAppInfoMetric(index, chain, txCache, internalState, metrics); err != nil {
		glog.Error("blockbookAppInfoMetric ", err)
	}

	var internalServer *server.InternalServer
	if *internalBinding != "" {
		internalServer, err = startInternalServer()
		if err != nil {
			glog.Error("internal server: ", err)
			return exitCodeFatal
		}
	}

	var publicServer *server.PublicServer
	if *publicBinding != "" {
		publicServer, err = startPublicServer()
		if err != nil {
			glog.Error("public server: ", err)
			return exitCodeFatal
		}
	}

	if *synchronize {
		internalState.SyncMode = true
		internalState.SetInitialSync(true)
		refreshSyncMetrics()
		// Keep the sync gauges live during the initial build/reindex. It can run for hours
		// to days, and syncIndexLoop and storeInternalStateLoop - the usual refreshers -
		// are only started after it returns, so without this ticker the single sample
		// above would be the only one taken during the whole initial sync and a stall
		// partway through would be invisible on initial_sync and synchronized.
		stopInitialRefresh := make(chan struct{})
		go func() {
			t := time.NewTicker(30 * time.Second)
			defer t.Stop()
			for {
				select {
				case <-stopInitialRefresh:
					return
				case <-t.C:
					refreshSyncMetrics()
				}
			}
		}()
		err := syncWorker.ResyncIndex(nil, true)
		close(stopInitialRefresh)
		if err != nil {
			if err != db.ErrOperationInterrupted {
				glog.Error("resyncIndex ", err)
				return exitCodeFatal
			}
			return exitCodeOK
		}
		// initialize mempool after the initial sync is complete
		var addrDescForOutpoint bchain.AddrDescForOutpointFunc
		if chain.GetChainParser().GetChainType() == bchain.ChainBitcoinType {
			addrDescForOutpoint = index.AddrDescForOutpoint
		}
		err = chain.InitializeMempool(addrDescForOutpoint, onNewTx)
		if err != nil {
			glog.Error("initializeMempool ", err)
			return exitCodeFatal
		}
		var mempoolCount int
		if mempoolCount, err = mempool.Resync(); err != nil {
			glog.Error("resyncMempool ", err)
			return exitCodeFatal
		}
		internalState.FinishedMempoolSync(mempoolCount)
		go syncIndexLoop()
		go syncMempoolLoop()
		go healInternalDataLoop()
		internalState.SetInitialSync(false)
		refreshSyncMetrics()
	}
	go storeInternalStateLoop()

	if publicServer != nil {
		// start full public interface
		callbacksOnNewBlock = append(callbacksOnNewBlock, publicServer.OnNewBlock)
		callbacksOnNewTx = append(callbacksOnNewTx, publicServer.OnNewTx)
		callbacksOnNewFiatRatesTicker = append(callbacksOnNewFiatRatesTicker, publicServer.OnNewFiatRatesTicker)
		publicServer.ConnectFullPublicInterface()
	}

	if *blockFrom >= 0 {
		if *blockUntil < 0 {
			*blockUntil = *blockFrom
		}
		height := uint32(*blockFrom)
		until := uint32(*blockUntil)

		if !*synchronize {
			if err = syncWorker.BulkConnectBlocks(height, until); err != nil {
				if err != db.ErrOperationInterrupted {
					glog.Error("connectBlocksParallel ", err)
					return exitCodeFatal
				}
				return exitCodeOK
			}
		}
	}

	if internalServer != nil || publicServer != nil || chain != nil {
		// start fiat rates downloader only if not shutting down immediately
		initDownloaders(index, chain, config)
		waitForSignalAndShutdown(internalServer, publicServer, chain, shutdownSigCh, 10*time.Second)
	}

	// Always stop periodic state storage to prevent writes during shutdown.
	close(chanStoreInternalState)
	if *synchronize {
		close(chanSyncIndex)
		close(chanSyncMempool)
		<-chanSyncIndexDone
		<-chanSyncMempoolDone
		// a healing pass writes to rocksdb from its own goroutine, so it has to be joined
		// here, before the deferred index.Close destroys the column family handles
		<-chanHealInternalDataDone
	}
	<-chanStoreInternalStateDone
	return exitCodeOK
}

func getBlockChainWithRetry(coin string, configFile string, pushHandler func(bchain.NotificationType), metrics *common.Metrics, seconds int) (bchain.BlockChain, bchain.Mempool, error) {
	var chain bchain.BlockChain
	var mempool bchain.Mempool
	var err error
	timer := time.NewTimer(time.Second)
	for i := 0; ; i++ {
		if chain, mempool, err = coins.NewBlockChain(coin, configFile, pushHandler, metrics); err != nil {
			if i < seconds {
				glog.Error("rpc: ", err, " Retrying...")
				select {
				case <-chanOsSignal:
					return nil, nil, errors.New("Interrupted")
				case <-timer.C:
					timer.Reset(time.Second)
					continue
				}
			} else {
				return nil, nil, err
			}
		}
		return chain, mempool, nil
	}
}

func startInternalServer() (*server.InternalServer, error) {
	internalServer, err := server.NewInternalServer(*internalBinding, *certFiles, index, chain, mempool, txCache, metrics, internalState, fiatRates)
	if err != nil {
		return nil, err
	}
	go func() {
		err = internalServer.Run()
		if err != nil {
			if err.Error() == "http: Server closed" {
				glog.Info("internal server: closed")
			} else {
				glog.Error(err)
				return
			}
		}
	}()
	return internalServer, nil
}

func startPublicServer() (*server.PublicServer, error) {
	// start public server in limited functionality, extend it after sync is finished by calling ConnectFullPublicInterface
	publicServer, err := server.NewPublicServer(*publicBinding, *certFiles, index, chain, mempool, txCache, *explorerURL, metrics, internalState, fiatRates, *debugMode)
	if err != nil {
		return nil, err
	}
	go func() {
		err = publicServer.Run()
		if err != nil {
			if err.Error() == "http: Server closed" {
				glog.Info("public server: closed")
			} else {
				glog.Error(err)
				return
			}
		}
	}()
	return publicServer, err
}

func performRollback() error {
	bestHeight, bestHash, err := index.GetBestBlock()
	if err != nil {
		glog.Error("rollbackHeight: ", err)
		return err
	}
	if uint32(*rollbackHeight) > bestHeight {
		glog.Infof("nothing to rollback, rollbackHeight %d, bestHeight: %d", *rollbackHeight, bestHeight)
	} else {
		hashes := []string{bestHash}
		for height := bestHeight - 1; height >= uint32(*rollbackHeight); height-- {
			hash, err := index.GetBlockHash(height)
			if err != nil {
				glog.Error("rollbackHeight: ", err)
				return err
			}
			hashes = append(hashes, hash)
		}
		err = syncWorker.DisconnectBlocks(uint32(*rollbackHeight), bestHeight, hashes)
		if err != nil {
			glog.Error("rollbackHeight: ", err)
			return err
		}
	}
	return nil
}

// refreshSyncMetrics publishes the sync-state gauges (synchronized, initial_sync and both
// heights) from internal state, with no backend round trip. It is called on every
// storeInternalStateLoop tick (~60s), after every sync iteration, at the two points
// is.InitialSync flips, on a ~30s ticker during the initial build (before those loops
// start), and from the app-info loop, so none of these gauges depends on the ~15-minute
// app-info loop for its cadence.
func refreshSyncMetrics() {
	if chain == nil {
		return
	}
	api.RefreshSyncMetrics(internalState, chain, metrics)
}

func blockbookAppInfoMetric(db *db.RocksDB, chain bchain.BlockChain, txCache *db.TxCache, is *common.InternalState, metrics *common.Metrics) error {
	// not named `api`: that shadowed the package for the rest of the function.
	w, err := api.NewWorker(db, chain, mempool, txCache, metrics, is, fiatRates)
	if err != nil {
		return err
	}
	si, err := w.GetSystemInfo(false)
	if err != nil {
		return err
	}
	// app_info carries purely local blockbook_* labels (version/commit/buildtime) that
	// identify the instance and need no backend, so the series must be published even when
	// the backend is unreachable - that is exactly the instance an operator needs to find.
	// On a backend error the backend_* labels and the height fall back to the last known
	// good BackendInfo this build retains, instead of blanking the whole series to zero.
	backend := *si.Backend // value copy: si.Backend is a pointer, don't mutate it in place
	if backend.BackendError != "" {
		bi := is.GetBackendInfo()
		backend.Version = bi.Version
		backend.Subversion = bi.Subversion
		backend.ConsensusVersion = bi.ConsensusVersion
		backend.ProtocolVersion = bi.ProtocolVersion
		backend.Blocks = bi.Blocks
	}
	subversion := backend.Subversion
	if subversion == "" {
		// for coins without subversion (ETH) use ConsensusVersion as subversion in metrics
		subversion = backend.ConsensusVersion
	}
	metrics.BlockbookAppInfo.Reset()
	metrics.BlockbookAppInfo.With(common.Labels{
		"blockbook_version":        si.Blockbook.Version,
		"blockbook_commit":         si.Blockbook.GitCommit,
		"blockbook_buildtime":      si.Blockbook.BuildTime,
		"backend_version":          backend.Version,
		"backend_subversion":       subversion,
		"backend_protocol_version": backend.ProtocolVersion}).Set(float64(0))
	if backend.Blocks > 0 {
		metrics.BackendBestHeight.Set(float64(backend.Blocks))
	}
	metrics.BackendTipAgeSeconds.Set(time.Since(is.GetBackendTipLastAdvance()).Seconds())
	// blockbook_synchronized, blockbook_best_height and blockbook_backend_best_height are
	// owned by RefreshSyncMetrics (called from the sync loop and the ~60s state loop).
	// Writing blockbook_synchronized here as well made two writers disagree - this path
	// derived it from a live GetChainInfo tip while RefreshSyncMetrics uses the cached one
	// - so the gauge oscillated on the app-info period. Publish it from the single shared
	// path instead of recomputing it here.
	api.RefreshSyncMetrics(is, chain, metrics)
	return nil
}

func newInternalState(config *common.Config, d *db.RocksDB, enableSubNewTx bool) (*common.InternalState, error) {
	is, err := d.LoadInternalState(config)
	if err != nil {
		return nil, err
	}

	is.EnableSubNewTx = enableSubNewTx
	name, err := os.Hostname()
	if err != nil {
		glog.Error("get hostname ", err)
	} else {
		if i := strings.IndexByte(name, '.'); i > 0 {
			name = name[:i]
		}
		is.Host = name
	}

	limitStr := os.Getenv(strings.ToUpper(is.GetNetwork()) + "_WS_GETACCOUNTINFO_LIMIT")
	if limitStr == "" {
		is.WsGetAccountInfoLimit = 42
	} else {
		is.WsGetAccountInfoLimit, _ = strconv.Atoi(limitStr)
	}
	if is.WsGetAccountInfoLimit > 0 {
		glog.Info("WsGetAccountInfoLimit enabled with limit ", is.WsGetAccountInfoLimit)
		is.WsLimitExceedingIPs = make(map[string]int)
	}

	// Balance-history transaction cap, split by transport. The shared
	// <NET>_BALANCE_HISTORY_MAX_TXS is a backward-compatible fallback that sets
	// the default for both surfaces; the transport-specific
	// <NET>_WS_BALANCE_HISTORY_MAX_TXS / <NET>_REST_BALANCE_HISTORY_MAX_TXS then
	// override their respective surface.
	netUpper := strings.ToUpper(is.GetNetwork())
	parseMaxTxs := func(envVar string, def int) (int, error) {
		if v, ok := os.LookupEnv(netUpper + envVar); ok {
			n, err := strconv.Atoi(strings.TrimSpace(v))
			if err != nil || n < 0 {
				return 0, errors.Errorf("%s%s: invalid value %q (want a non-negative integer, 0 to disable)", netUpper, envVar, v)
			}
			return n, nil
		}
		return def, nil
	}
	wsDefault, restDefault := api.DefaultBalanceHistoryMaxTxsWS, api.DefaultBalanceHistoryMaxTxsREST
	if shared, err := parseMaxTxs("_BALANCE_HISTORY_MAX_TXS", -1); err != nil {
		return nil, err
	} else if shared >= 0 {
		wsDefault, restDefault = shared, shared
	}
	if is.BalanceHistoryMaxTxsWS, err = parseMaxTxs("_WS_BALANCE_HISTORY_MAX_TXS", wsDefault); err != nil {
		return nil, err
	}
	if is.BalanceHistoryMaxTxsREST, err = parseMaxTxs("_REST_BALANCE_HISTORY_MAX_TXS", restDefault); err != nil {
		return nil, err
	}
	logMaxTxs := func(surface string, n int) {
		if n > 0 {
			glog.Info("BalanceHistoryMaxTxs ", surface, " limit ", n, " transactions per request")
		} else {
			glog.Info("BalanceHistoryMaxTxs ", surface, " unlimited")
		}
	}
	logMaxTxs("WS", is.BalanceHistoryMaxTxsWS)
	logMaxTxs("REST", is.BalanceHistoryMaxTxsREST)
	return is, nil
}

func syncIndexLoop() {
	defer close(chanSyncIndexDone)
	glog.Info("syncIndexLoop starting")
	// resync index about every 15 minutes if there are no chanSyncIndex requests, with debounce 1 second
	common.TickAndDebounce(time.Duration(*resyncIndexPeriodMs)*time.Millisecond, time.Duration(*resyncIndexDebounceMs)*time.Millisecond, chanSyncIndex, func() {
		// Publish the sync-state gauges after every iteration whatever its outcome. The
		// early returns below make a defer the only placement that covers all of them,
		// and the interesting case is precisely the quiet one: during a silent stall
		// resyncIndex keeps returning syncNotNeeded, so nothing else would move the
		// gauges until the ~60s storeInternalStateLoop tick.
		defer refreshSyncMetrics()
		if err := syncWorker.ResyncIndex(onNewBlock, false); err != nil {
			if err == db.ErrOperationInterrupted || common.IsInShutdown() {
				return
			}
			glog.Error("syncIndexLoop ", errors.ErrorStack(err), ", will retry...")
			// retry once in case of random network error, after a slight delay
			time.Sleep(time.Millisecond * 2500)
			if err := syncWorker.ResyncIndex(onNewBlock, false); err != nil {
				if err == db.ErrOperationInterrupted || common.IsInShutdown() {
					return
				}
				glog.Error("syncIndexLoop ", errors.ErrorStack(err))
			}
		}
	})
	glog.Info("syncIndexLoop stopped")
}

func onNewBlock(block *bchain.Block) {
	defer func() {
		if r := recover(); r != nil {
			glog.Error("onNewBlockHash recovered from panic: ", r)
		}
	}()
	for _, c := range callbacksOnNewBlock {
		c(block)
	}
}

func onNewFiatRatesTicker(ticker *common.CurrencyRatesTicker) {
	defer func() {
		if r := recover(); r != nil {
			glog.Error("onNewFiatRatesTicker recovered from panic: ", r)
			debug.PrintStack()
		}
	}()
	for _, c := range callbacksOnNewFiatRatesTicker {
		c(ticker)
	}
}

func syncMempoolLoop() {
	defer close(chanSyncMempoolDone)
	glog.Info("syncMempoolLoop starting")
	// resync mempool about every minute if there are no chanSyncMempool requests, with debounce 1 second
	common.TickAndDebounce(time.Duration(*resyncMempoolPeriodMs)*time.Millisecond, debounceResyncMempoolMs*time.Millisecond, chanSyncMempool, func() {
		internalState.StartedMempoolSync()
		if count, err := mempool.Resync(); err != nil {
			glog.Error("syncMempoolLoop ", errors.ErrorStack(err))
		} else {
			internalState.FinishedMempoolSync(count)

		}
	})
	glog.Info("syncMempoolLoop stopped")
}

// healInternalDataLoop drains the internal data error queue without operator action:
// blocks whose internal data failed during sync are refetched and reconnected to the
// index. It returns on shutdown; the caller joins it before the database is closed.
func healInternalDataLoop() {
	defer close(chanHealInternalDataDone)
	// nothing can be queued unless internal data are processed on an ethereum-type chain
	if chain.GetChainParser().GetChainType() != bchain.ChainEthereumType || !bchain.ProcessInternalTransactions {
		return
	}
	w, err := api.NewWorker(index, chain, mempool, txCache, metrics, internalState, fiatRates)
	if err != nil {
		glog.Error("healInternalDataLoop: NewWorker ", err)
		return
	}
	glog.Info("healInternalDataLoop starting")
	defer glog.Info("healInternalDataLoop stopped")
	timer := time.NewTimer(healInternalDataStartupDelay)
	defer timer.Stop()
	for pass := uint64(0); ; pass++ {
		select {
		// closed on shutdown, so this is a broadcast to every waiting loop
		case <-chanOsSignal:
			return
		case <-timer.C:
		}
		timer.Reset(healInternalDataPeriod)
		if common.IsInShutdown() {
			return
		}
		w.HealInternalData(pass, chanOsSignal)
	}
}

func storeInternalStateLoop() {
	defer func() {
		close(chanStoreInternalStateDone)
	}()
	// Reuse the global shutdown channel so compute work stops when shutdown begins.
	stopCompute := chanOsSignal
	var computeRunning bool
	lastCompute := time.Now()
	lastAppInfo := time.Now()
	logAppInfoPeriod := 15 * time.Minute
	// randomize the duration between ComputeInternalStateColumnStats to avoid peaks after reboot of machine with multiple blockbooks
	computePeriod := time.Duration(*dbStatsPeriodHours)*time.Hour + time.Duration(rand.Float64()*float64((4*time.Hour).Nanoseconds()))
	if (*dbStatsPeriodHours) > 0 {
		glog.Info("storeInternalStateLoop starting with db stats recompute period ", computePeriod)
	} else {
		glog.Info("storeInternalStateLoop starting with db stats compute disabled")
	}
	common.TickAndDebounce(storeInternalStatePeriodMs*time.Millisecond, (storeInternalStatePeriodMs-1)*time.Millisecond, chanStoreInternalState, func() {
		// Republish the sync-state gauges every tick (~60s) - the guaranteed freshness
		// floor. syncIndexLoop also refreshes them, but a blockbook whose tip feed went
		// quiet only wakes on the resyncindexperiod backstop (~15.6 min), which is how far
		// blockbook_synchronized used to lag reality. RefreshSyncMetrics reads internal
		// state only, no backend RPCs, so it is cheap to run unconditionally.
		refreshSyncMetrics()
		if (*dbStatsPeriodHours) > 0 && !computeRunning && lastCompute.Add(computePeriod).Before(time.Now()) {
			computeRunning = true
			go func() {
				err := index.ComputeInternalStateColumnStats(stopCompute)
				if err != nil {
					glog.Error("computeInternalStateColumnStats error: ", err)
				}
				lastCompute = time.Now()
				computeRunning = false
			}()
		}
		if err := index.StoreInternalState(internalState); err != nil {
			glog.Error("storeInternalStateLoop ", errors.ErrorStack(err))
		}
		if lastAppInfo.Add(logAppInfoPeriod).Before(time.Now()) {
			if glog.V(1) {
				glog.Info(index.GetMemoryStats())
			}
			if err := blockbookAppInfoMetric(index, chain, txCache, internalState, metrics); err != nil {
				glog.Error("blockbookAppInfoMetric ", err)
			}
			lastAppInfo = time.Now()
		}
	})
	glog.Info("storeInternalStateLoop stopped")
}

func onNewTx(tx *bchain.MempoolTx) {
	defer func() {
		if r := recover(); r != nil {
			glog.Error("onNewTx recovered from panic: ", r)
		}
	}()
	for _, c := range callbacksOnNewTx {
		c(tx)
	}
}

func pushSynchronizationHandler(nt bchain.NotificationType) {
	glog.V(1).Info("MQ: notification ", nt)
	if common.IsInShutdown() {
		return
	}
	if nt == bchain.NotificationNewBlock {
		select {
		case chanSyncIndex <- struct{}{}:
		default:
		}
	} else if nt == bchain.NotificationNewTx {
		select {
		case chanSyncMempool <- struct{}{}:
		default:
		}
	} else {
		glog.Error("MQ: unknown notification sent")
	}
}

func waitForSignalAndShutdown(internal *server.InternalServer, public *server.PublicServer, chain bchain.BlockChain, shutdownSig <-chan os.Signal, timeout time.Duration) {
	// Read the first OS signal from the dedicated channel to avoid races with worker shutdown paths.
	sig := <-shutdownSig
	common.SetInShutdown()
	if sig != nil {
		glog.Infof("shutdown: %v", sig)
	} else {
		glog.Info("shutdown: signal received")
	}

	// Bound server/RPC shutdown; RocksDB close happens after main returns via defer.
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()

	if internal != nil {
		if err := internal.Shutdown(ctx); err != nil {
			glog.Error("internal server: shutdown error: ", err)
		}
	}

	if public != nil {
		if err := public.Shutdown(ctx); err != nil {
			glog.Error("public server: shutdown error: ", err)
		}
	}

	if chain != nil {
		if err := chain.Shutdown(ctx); err != nil {
			glog.Error("rpc: shutdown error: ", err)
		}
	}
}

// computeFeeStats computes fee distribution in defined blocks
func computeFeeStats(stopCompute chan os.Signal, blockFrom, blockTo int, db *db.RocksDB, chain bchain.BlockChain, txCache *db.TxCache, is *common.InternalState, metrics *common.Metrics) error {
	start := time.Now()
	glog.Info("computeFeeStats start")
	api, err := api.NewWorker(db, chain, mempool, txCache, metrics, is, fiatRates)
	if err != nil {
		return err
	}
	err = api.ComputeFeeStats(blockFrom, blockTo, stopCompute)
	glog.Info("computeFeeStats finished in ", time.Since(start))
	return err
}

// runStartupSelfHealing runs blocking, at-startup self-healing tasks once RocksDB is
// initialized. It blocks until done so the rest of startup proceeds against a consistent DB.
// Future self-healing tasks can be added here; today it reconciles missing historical fiat
// rates before the periodic downloader loops start.
func runStartupSelfHealing() {
	if fiatRates != nil && fiatRates.Enabled {
		// chanOsSignal is closed on shutdown, so a SIGTERM during a long reconciliation
		// aborts it promptly instead of blocking startup-shutdown until the backfill ends.
		fiatRates.ReconcileHistoricalRatesAtStartup(chanOsSignal)
	}
}

func initDownloaders(db *db.RocksDB, chain bchain.BlockChain, config *common.Config) {
	if fiatRates.Enabled {
		// Block on startup self-healing before the periodic loops begin, so historical gaps
		// are repaired via the CDN without contending with the Free-tier tip loop.
		runStartupSelfHealing()
		go fiatRates.RunDownloader()
	}

	if config.FourByteSignatures != "" && chain.GetChainParser().GetChainType() == bchain.ChainEthereumType {
		fbsd, err := fourbyte.NewFourByteSignaturesDownloader(db, config.FourByteSignatures)
		if err != nil {
			glog.Errorf("NewFourByteSignaturesDownloader Init error: %v", err)
		} else {
			glog.Infof("Starting FourByteSignatures downloader...")
			go fbsd.Run()
		}

	}

}