Repository: algorand/indexer
Pull request #587 (2021-07-23 15:41:04)

Add more prometheus metrics.
## Summary

Added some new metrics to go along with `indexer_daemon_import_time_sec`:
-  block_upload_time_sec
-  imported_tx_per_block
-  imported_round

Also moved the location where metrics are collected, previously they did not represent exactly one round worth of data.

Updated the block generator import test runner to capture the new metrics.

New report output
```
cat cmd/block-generator/results/config.asset.close.report
test_duration_seconds:3600
test_duration_actual_seconds:3600.027172
transaction_asset_create_total:2134
transaction_asset_optin_total:1065726
transaction_asset_xfer_total:208325
transaction_asset_close_total:843815
early_average_import_time_sec:8.17
early_cumulative_import_time_sec:1054.50
early_average_imported_tx_per_block:5000.00
early_cumulative_imported_tx_per_block:645000
early_average_block_upload_time_sec:0.36
early_cumulative_block_upload_time_sec:47.54
early_imported_round:130
early_overall_transactions_per_second:611.67
early_uptime_seconds:3600.03
final_average_import_time_sec:8.42
final_cumulative_import_time_sec:3554.10
final_average_imported_tx_per_block:5000.00
final_cumulative_imported_tx_per_block:2110000
final_average_block_upload_time_sec:0.37
final_cumulative_block_upload_time_sec:155.84
final_imported_round:423
final_overall_transactions_per_second:593.68
final_uptime_seconds:3600.03
```

## Test Plan

Run the test runner and observed that the metrics look about right.


codecov-commenter: # [Codecov](https://codecov.io/gh/algorand/indexer/pull/587?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) Report
> Merging [#587](https://codecov.io/gh/algorand/indexer/pull/587?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (3a05bdf) into [develop](https://codecov.io/gh/algorand/indexer/commit/6e4d737f2e4e49088b436a234caee6681435053d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (6e4d737) will **not change** coverage.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/algorand/indexer/pull/587/graphs/tree.svg?width=650&height=150&src=pr&token=BSHI3PT6QP&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)](https://codecov.io/gh/algorand/indexer/pull/587?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)

```diff
@@           Coverage Diff            @@
##           develop     #587   +/-   ##
========================================
  Coverage    47.50%   47.50%           
========================================
  Files           25       25           
  Lines         3846     3846           
========================================
  Hits          1827     1827           
  Misses        1739     1739           
  Partials       280      280           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/algorand/indexer/pull/587?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/algorand/indexer/pull/587?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None). Last update [6e4d737...3a05bdf](https://codecov.io/gh/algorand/indexer/pull/587?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None).


Patch:
diff --git a/Makefile b/Makefile
index a0570af..94bc871 100644
--- a/Makefile
+++ b/Makefile
@@ -1,7 +1,7 @@
 SRCPATH		:= $(shell pwd)
 VERSION		:= $(shell $(SRCPATH)/mule/scripts/compute_build_number.sh)
-OS_TYPE		:= $(shell $(SRCPATH)/mule/scripts/ostype.sh)
-ARCH			:= $(shell $(SRCPATH)/mule/scripts/archtype.sh)
+OS_TYPE		?= $(shell $(SRCPATH)/mule/scripts/ostype.sh)
+ARCH			?= $(shell $(SRCPATH)/mule/scripts/archtype.sh)
 PKG_DIR		= $(SRCPATH)/tmp/node_pkgs/$(OS_TYPE)/$(ARCH)/$(VERSION)
 
 # TODO: ensure any additions here are mirrored in misc/release.py
diff --git a/README.md b/README.md
index c2642ce..ab949b6 100644
--- a/README.md
+++ b/README.md
@@ -67,7 +67,7 @@ The most common usage of the Indexer is expecting to be getting validated blocks
 
 Indexer works by fetching blocks one at a time, processing the block data, and loading it into a traditional database. There is a database abstraction layer to support different database implementations. In normal operation, the service will run as a daemon and always requires access to a database.
 
-As of April 2020, storing all the raw blocks is about 100 GB and the PostgreSQL database of transactions and accounts is about 1 GB. Much of that size difference is the Indexer ignoring cryptographic signature data; relying on `algod` to validate blocks. Dropping that, the Indexer can focus on the 'what happened' details of transactions and accounts.
+As of the end of July 2021, storing all the raw blocks in MainNet is about 609 GB and the PostgreSQL database of transactions and accounts is about 495 GB. Much of that size difference is the Indexer ignoring cryptographic signature data; relying on `algod` to validate blocks. Dropping that, the Indexer can focus on the 'what happened' details of transactions and accounts.
 
 There are two primary modes of operation:
 * [Database updater](#database-updater)
diff --git a/cmd/algorand-indexer/daemon.go b/cmd/algorand-indexer/daemon.go
index 8fb6e39..fd31504 100644
--- a/cmd/algorand-indexer/daemon.go
+++ b/cmd/algorand-indexer/daemon.go
@@ -7,7 +7,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/prometheus/client_golang/prometheus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 
@@ -17,6 +16,7 @@ import (
 	"github.com/algorand/indexer/idb"
 	"github.com/algorand/indexer/importer"
 	"github.com/algorand/indexer/types"
+	"github.com/algorand/indexer/util/metrics"
 )
 
 var (
@@ -31,23 +31,12 @@ var (
 	tokenString      string
 )
 
-// importTimeHistogramSeconds is used to record the block import time metric.
-var importTimeHistogramSeconds = prometheus.NewSummary(
-	prometheus.SummaryOpts{
-		Subsystem: "indexer_daemon",
-		Name:      "import_time_sec",
-		Help:      "Block import and processing time in seconds.",
-	})
-
 var daemonCmd = &cobra.Command{
 	Use:   "daemon",
 	Short: "run indexer daemon",
 	Long:  "run indexer daemon. Serve api on HTTP.",
 	//Args:
 	Run: func(cmd *cobra.Command, args []string) {
-		// register metric with global prometheus metrics handler
-		prometheus.Register(importTimeHistogramSeconds)
-
 		var err error
 		config.BindFlags(cmd)
 		err = configureLogger()
@@ -79,30 +68,31 @@ var daemonCmd = &cobra.Command{
 		if noAlgod && !allowMigration {
 			opts.ReadOnly = true
 		}
-		db := indexerDbFromFlags(opts)
+		db, availableCh := indexerDbFromFlags(opts)
 		if bot != nil {
-			logger.Info("Initializing block import handler.")
+			go func() {
+				// Wait until the database is available.
+				<-availableCh
 
-			nextRound, err := db.GetNextRoundToLoad()
-			maybeFail(err, "failed to get next round, %v", err)
-			bot.SetNextRound(nextRound)
+				// Initial import if needed.
+				importer.InitialImport(db, genesisJSONPath, bot.Algod(), logger)
 
-			cache, err := db.GetDefaultFrozen()
-			maybeFail(err, "failed to get default frozen cache")
+				logger.Info("Initializing block import handler.")
 
-			bih := blockImporterHandler{
-				imp:   importer.NewDBImporter(db),
-				db:    db,
-				cache: cache,
-			}
-			bot.AddBlockHandler(&bih)
-			bot.SetContext(ctx)
+				nextRound, err := db.GetNextRoundToLoad()
+				maybeFail(err, "failed to get next round, %v", err)
+				bot.SetNextRound(nextRound)
 
-			go func() {
-				waitForDBAvailable(db)
+				cache, err := db.GetDefaultFrozen()
+				maybeFail(err, "failed to get default frozen cache")
 
-				// Initial import if needed.
-				importer.InitialImport(db, genesisJSONPath, bot.Algod(), logger)
+				bih := blockImporterHandler{
+					imp:   importer.NewDBImporter(db),
+					db:    db,
+					cache: cache,
+				}
+				bot.AddBlockHandler(&bih)
+				bot.SetContext(ctx)
 
 				logger.Info("Starting block importer.")
 				bot.Run()
@@ -119,35 +109,6 @@ var daemonCmd = &cobra.Command{
 	},
 }
 
-// waitForDBAvailable wait for the IndexerDb to report that it is available.
-func waitForDBAvailable(db idb.IndexerDb) {
-	statusInterval := 5 * time.Minute
-	checkInterval := 5 * time.Second
-	var now time.Time
-	nextStatusTime := time.Now()
-	for true {
-		now = time.Now()
-		health, err := db.Health()
-		if err != nil {
-			logger.WithError(err).Errorf("Problem fetching database health.")
-			os.Exit(1)
-		}
-
-		// Exit function when the database is available
-		if health.DBAvailable {
-			return
-		}
-
-		// Log status periodically
-		if nextStatusTime.Sub(now) <= 0 {
-			logger.Info("Block importer waiting for database to become available.")
-			nextStatusTime = nextStatusTime.Add(statusInterval)
-		}
-
-		time.Sleep(checkInterval)
-	}
-}
-
 func init() {
 	daemonCmd.Flags().StringVarP(&algodDataDir, "algod", "d", "", "path to algod data dir, or $ALGORAND_DATA")
 	daemonCmd.Flags().StringVarP(&algodAddr, "algod-net", "", "", "host:port of algod")
@@ -197,6 +158,7 @@ func (bih *blockImporterHandler) HandleBlock(block *types.EncodedBlockCert) {
 	start := time.Now()
 	_, err := bih.imp.ImportDecodedBlock(block)
 	maybeFail(err, "ImportDecodedBlock %d", block.Block.Round)
+	metrics.BlockUploadTimeSeconds.Observe(time.Since(start).Seconds())
 	startRound, err := bih.db.GetNextRoundToAccount()
 	maybeFail(err, "failed to get next round to account")
 	// During normal operation StartRound and MaxRound will be the same round.
@@ -204,9 +166,15 @@ func (bih *blockImporterHandler) HandleBlock(block *types.EncodedBlockCert) {
 		StartRound: startRound,
 		MaxRound:   uint64(block.Block.Round),
 	}
-	importer.UpdateAccounting(bih.db, bih.cache, filter, logger)
-	dt := time.Now().Sub(start)
-	// record metric
-	importTimeHistogramSeconds.Observe(dt.Seconds())
+	rounds, txns := importer.UpdateAccounting(bih.db, bih.cache, filter, logger)
+	dt := time.Since(start)
+
+	// Ignore calls that update >1 round (sneaky migration) and round 0 (which is empty)
+	if rounds <= 1 && startRound != 0 {
+		metrics.BlockImportTimeSeconds.Observe(dt.Seconds())
+		metrics.ImportedTxnsPerBlock.Observe(float64(txns))
+		metrics.ImportedRoundGauge.Set(float64(startRound))
+	}
+
 	logger.Infof("round r=%d (%d txn) imported in %s", block.Block.Round, len(block.Block.Payset), dt.String())
 }
diff --git a/cmd/algorand-indexer/import.go b/cmd/algorand-indexer/import.go
index 7d8f1bd..16bbf18 100644
--- a/cmd/algorand-indexer/import.go
+++ b/cmd/algorand-indexer/import.go
@@ -23,7 +23,8 @@ var importCmd = &cobra.Command{
 			os.Exit(1)
 		}
 
-		db := indexerDbFromFlags(idb.IndexerDbOptions{})
+		db, availableCh := indexerDbFromFlags(idb.IndexerDbOptions{})
+		<-availableCh
 
 		cache, err := db.GetDefaultFrozen()
 		if err != nil {
diff --git a/cmd/algorand-indexer/main.go b/cmd/algorand-indexer/main.go
index 5211711..c3a32c8 100644
--- a/cmd/algorand-indexer/main.go
+++ b/cmd/algorand-indexer/main.go
@@ -16,6 +16,7 @@ import (
 	"github.com/algorand/indexer/idb"
 	"github.com/algorand/indexer/idb/dummy"
 	_ "github.com/algorand/indexer/idb/postgres"
+	"github.com/algorand/indexer/util/metrics"
 	"github.com/algorand/indexer/version"
 )
 
@@ -84,18 +85,18 @@ var (
 	logger         *log.Logger
 )
 
-func indexerDbFromFlags(opts idb.IndexerDbOptions) (db idb.IndexerDb) {
+func indexerDbFromFlags(opts idb.IndexerDbOptions) (idb.IndexerDb, chan struct{}) {
 	if postgresAddr != "" {
-		var err error
-		db, err = idb.IndexerDbByName("postgres", postgresAddr, opts, logger)
+		db, ch, err := idb.IndexerDbByName("postgres", postgresAddr, opts, logger)
 		maybeFail(err, "could not init db, %v", err)
-	} else if dummyIndexerDb {
-		db = dummy.IndexerDb()
-	} else {
-		logger.Errorf("no import db set")
-		os.Exit(1)
+		return db, ch
+	}
+	if dummyIndexerDb {
+		return dummy.IndexerDb(), nil
 	}
-	return db
+	logger.Errorf("no import db set")
+	os.Exit(1)
+	return nil, nil
 }
 
 func init() {
@@ -141,6 +142,9 @@ func init() {
 
 	viper.SetEnvPrefix(config.EnvPrefix)
 	viper.AutomaticEnv()
+
+	// Register metrics with the global prometheus handler.
+	metrics.RegisterPrometheusMetrics()
 }
 
 func configureLogger() error {
diff --git a/cmd/block-generator/runner/metrics_collector.go b/cmd/block-generator/runner/metrics_collector.go
index ccaf545..7c89aed 100644
--- a/cmd/block-generator/runner/metrics_collector.go
+++ b/cmd/block-generator/runner/metrics_collector.go
@@ -23,8 +23,8 @@ type Entry struct {
 }
 
 // Collect fetches the metrics.
-func (r *MetricsCollector) Collect(prefix string) error {
-	metrics, err := r.getMetrics(prefix)
+func (r *MetricsCollector) Collect(substrings ...string) error {
+	metrics, err := r.getMetrics(substrings...)
 	if err != nil {
 		return err
 	}
@@ -40,7 +40,7 @@ func (r *MetricsCollector) Collect(prefix string) error {
 	return nil
 }
 
-func (r MetricsCollector) getMetrics(prefix string) (result []string, err error) {
+func (r MetricsCollector) getMetrics(substrings ...string) (result []string, err error) {
 	resp, err := http.Get(r.MetricsURL)
 	if err != nil {
 		err = fmt.Errorf("unable to read metrics url '%s'", r.MetricsURL)
@@ -56,8 +56,11 @@ func (r MetricsCollector) getMetrics(prefix string) (result []string, err error)
 			continue
 		}
 
-		if strings.HasPrefix(str, prefix) {
-			result = append(result, str)
+		for _, substring := range substrings {
+			if strings.Contains(str, substring) {
+				result = append(result, str)
+				break
+			}
 		}
 	}
 
diff --git a/cmd/block-generator/runner/run.go b/cmd/block-generator/runner/run.go
index 05d65df..39206af 100644
--- a/cmd/block-generator/runner/run.go
+++ b/cmd/block-generator/runner/run.go
@@ -20,6 +20,7 @@ import (
 
 	"github.com/algorand/indexer/cmd/block-generator/generator"
 	"github.com/algorand/indexer/util"
+	"github.com/algorand/indexer/util/metrics"
 )
 
 // Args are all the things needed to run a performance test.
@@ -82,33 +83,138 @@ func (r *Args) run() error {
 	return nil
 }
 
-// Helper to record the import rate.
-func recordDataToFile(entry Entry, key string, out *os.File) error {
-	sum := 0.0
-	count := 0.0
+type metricType int
 
-	for _, metric := range entry.Data {
-		var err error
-		if strings.HasPrefix(metric, "indexer_daemon_import_time_sec_sum") {
-			val := strings.Split(metric, " ")[1]
-			sum, err = strconv.ParseFloat(val, 64)
-		} else if strings.HasPrefix(metric, "indexer_daemon_import_time_sec_count") {
-			val := strings.Split(metric, " ")[1]
-			count, err = strconv.ParseFloat(val, 64)
-		}
-		if err != nil {
-			return fmt.Errorf("unable to parse metric '%s': %w", metric, err)
+const (
+	rate metricType = iota
+	intTotal
+	floatTotal
+)
+
+// Helper to record metrics. Supports rates (sum/count) and counters.
+func recordDataToFile(start time.Time, entry Entry, prefix string, out *os.File) error {
+	var writeErrors strings.Builder
+	var writeErr error
+	record := func(prefix2, name string, t metricType) {
+		key := fmt.Sprintf("%s%s_%s", prefix, prefix2, name)
+		if err := recordMetricToFile(entry, key, name, t, out); err != nil {
+			writeErr = err
+			if writeErrors.Len() > 0 {
+				writeErrors.WriteString(", ")
+			}
+			writeErrors.WriteString(name)
 		}
 	}
-	rate := sum / count
 
-	msg := fmt.Sprintf("%s:%f\n", key, rate)
+	record("_average", metrics.BlockImportTimeName, rate)
+	record("_cumulative", metrics.BlockImportTimeName, floatTotal)
+	record("_average", metrics.ImportedTxnsPerBlockName, rate)
+	record("_cumulative", metrics.ImportedTxnsPerBlockName, intTotal)
+	record("_average", metrics.BlockUploadTimeName, rate)
+	record("_cumulative", metrics.BlockUploadTimeName, floatTotal)
+	record("", metrics.ImportedRoundGaugeName, intTotal)
+
+	if writeErrors.Len() > 0 {
+		return fmt.Errorf("error writing metrics (%s): %w", writeErrors.String(), writeErr)
+	}
+
+	// Calculate import transactions per second.
+	totalTxn, err := getMetric(entry, metrics.ImportedTxnsPerBlockName, false)
+	if err != nil {
+		return err
+	}
+
+	importTimeS, err := getMetric(entry, metrics.BlockImportTimeName, false)
+	if err != nil {
+		return err
+	}
+	tps := totalTxn / importTimeS
+	key := "overall_transactions_per_second"
+	msg := fmt.Sprintf("%s_%s:%.2f\n", prefix, key, tps)
+	if _, err := out.WriteString(msg); err != nil {
+		return fmt.Errorf("unable to write metric '%s': %w", key, err)
+	}
+
+	// Uptime
+	key = "uptime_seconds"
+	msg = fmt.Sprintf("%s_%s:%.2f\n", prefix, key, time.Since(start).Seconds())
 	if _, err := out.WriteString(msg); err != nil {
 		return fmt.Errorf("unable to write metric '%s': %w", key, err)
 	}
+
 	return nil
 }
 
+func recordMetricToFile(entry Entry, outputKey, metricSuffix string, t metricType, out *os.File) error {
+	value, err := getMetric(entry, metricSuffix, t == rate)
+	if err != nil {
+		return err
+	}
+
+	var msg string
+	if t == intTotal {
+		msg = fmt.Sprintf("%s:%d\n", outputKey, uint64(value))
+	} else {
+		msg = fmt.Sprintf("%s:%.2f\n", outputKey, value)
+	}
+
+	if _, err := out.WriteString(msg); err != nil {
+		return fmt.Errorf("unable to write metric '%s': %w", outputKey, err)
+	}
+
+	return nil
+}
+
+func getMetric(entry Entry, suffix string, rateMetric bool) (float64, error) {
+	total := 0.0
+	sum := 0.0
+	count := 0.0
+	hasSum := false
+	hasCount := false
+	hasTotal := false
+
+	for _, metric := range entry.Data {
+		var err error
+
+		if strings.Contains(metric, suffix) {
+			split := strings.Split(metric, " ")
+			if len(split) != 2 {
+				return 0.0, fmt.Errorf("unknown metric format, expected 'key value' received: %s", metric)
+			}
+
+			// Check for _sum / _count for summary (rateMetric) metrics.
+			// Otherwise grab the total value.
+			if strings.HasSuffix(split[0], "_sum") {
+				sum, err = strconv.ParseFloat(split[1], 64)
+				hasSum = true
+			} else if strings.HasSuffix(split[0], "_count") {
+				count, err = strconv.ParseFloat(split[1], 64)
+				hasCount = true
+			} else if strings.HasSuffix(split[0], suffix) {
+				total, err = strconv.ParseFloat(split[1], 64)
+				hasTotal = true
+			}
+
+			if err != nil {
+				return 0.0, fmt.Errorf("unable to parse metric '%s': %w", metric, err)
+			}
+
+			if rateMetric && hasSum && hasCount {
+				return sum / count, nil
+			} else if !rateMetric {
+				if hasSum {
+					return sum, nil
+				}
+				if hasTotal {
+					return total, nil
+				}
+			}
+		}
+	}
+
+	return 0.0, fmt.Errorf("metric incomplete or not found: %s", suffix)
+}
+
 // Run the test for 'RunDuration', collect metrics and write them to the 'ReportDirectory'
 func (r *Args) runTest(indexerURL string, generatorURL string) error {
 	collector := &MetricsCollector{MetricsURL: fmt.Sprintf("http://%s/metrics", indexerURL)}
@@ -127,11 +233,12 @@ func (r *Args) runTest(indexerURL string, generatorURL string) error {
 	start := time.Now()
 	for time.Since(start) < r.RunDuration {
 		time.Sleep(r.RunDuration / 10)
-		if err := collector.Collect("indexer_daemon_import_time_sec"); err != nil {
+
+		if err := collector.Collect(metrics.AllMetricNames...); err != nil {
 			return fmt.Errorf("problem collecting metrics: %w", err)
 		}
 	}
-	if err := collector.Collect("indexer_daemon_import_time_sec"); err != nil {
+	if err := collector.Collect(metrics.AllMetricNames...); err != nil {
 		return fmt.Errorf("problem collecting metrics: %w", err)
 	}
 
@@ -165,13 +272,13 @@ func (r *Args) runTest(indexerURL string, generatorURL string) error {
 
 	// Record a rate from one of the first data points.
 	if len(collector.Data) > 5 {
-		if err := recordDataToFile(collector.Data[2], "starting_block_import_duration_average_seconds", report); err != nil {
+		if err := recordDataToFile(start, collector.Data[2], "early", report); err != nil {
 			return err
 		}
 	}
 
 	// Also record the final one.
-	if err := recordDataToFile(collector.Data[len(collector.Data)-1], "final_import_duration_average_seconds", report); err != nil {
+	if err := recordDataToFile(start, collector.Data[len(collector.Data)-1], "final", report); err != nil {
 		return err
 	}
 
diff --git a/cmd/validator/struct_processor.go b/cmd/validator/struct_processor.go
index aa79b4c..fce903e 100644
--- a/cmd/validator/struct_processor.go
+++ b/cmd/validator/struct_processor.go
@@ -183,6 +183,15 @@ func equals(indexer, algod generated.Account) (differences []string) {
 			if !stringPtrEqual(algodCreatedAsset.Params.Url, indexerCreatedAsset.Params.Url) {
 				differences = append(differences, fmt.Sprintf("created-asset url %d", algodCreatedAsset.Index))
 			}
+			if !bytesPtrEqual(algodCreatedAsset.Params.NameB64, indexerCreatedAsset.Params.NameB64) {
+				differences = append(differences, fmt.Sprintf("created-asset name-b64 %d", algodCreatedAsset.Index))
+			}
+			if !bytesPtrEqual(algodCreatedAsset.Params.UnitNameB64, indexerCreatedAsset.Params.UnitNameB64) {
+				differences = append(differences, fmt.Sprintf("created-asset unit-name-b64 %d", algodCreatedAsset.Index))
+			}
+			if !bytesPtrEqual(algodCreatedAsset.Params.UrlB64, indexerCreatedAsset.Params.UrlB64) {
+				differences = append(differences, fmt.Sprintf("created-asset url-b64 %d", algodCreatedAsset.Index))
+			}
 			if !bytesPtrEqual(algodCreatedAsset.Params.MetadataHash, indexerCreatedAsset.Params.MetadataHash) {
 				differences = append(differences, fmt.Sprintf("created-asset metadata-hash %d", algodCreatedAsset.Index))
 			}
diff --git a/idb/dummy/dummy_factory.go b/idb/dummy/dummy_factory.go
index 310dae3..c093a32 100644
--- a/idb/dummy/dummy_factory.go
+++ b/idb/dummy/dummy_factory.go
@@ -15,8 +15,8 @@ func (df dummyFactory) Name() string {
 }
 
 // Build is part of the IndexerFactory interface.
-func (df dummyFactory) Build(arg string, opts idb.IndexerDbOptions, log *log.Logger) (idb.IndexerDb, error) {
-	return &dummyIndexerDb{log: log}, nil
+func (df dummyFactory) Build(arg string, opts idb.IndexerDbOptions, log *log.Logger) (idb.IndexerDb, chan struct{}, error) {
+	return &dummyIndexerDb{log: log}, nil, nil
 }
 
 func init() {
diff --git a/idb/idb_factory.go b/idb/idb_factory.go
index b67a30a..1ea221a 100644
--- a/idb/idb_factory.go
+++ b/idb/idb_factory.go
@@ -9,7 +9,7 @@ import (
 // IndexerDbFactory is used to install an IndexerDb implementation.
 type IndexerDbFactory interface {
 	Name() string
-	Build(arg string, opts IndexerDbOptions, log *log.Logger) (IndexerDb, error)
+	Build(arg string, opts IndexerDbOptions, log *log.Logger) (IndexerDb, chan struct{}, error)
 }
 
 // This layer of indirection allows for different db integrations to be compiled in or compiled out by `go build --tags ...`
@@ -23,11 +23,13 @@ func RegisterFactory(name string, factory IndexerDbFactory) {
 }
 
 // IndexerDbByName is used to construct an IndexerDb object by name.
-func IndexerDbByName(name, arg string, opts IndexerDbOptions, log *log.Logger) (IndexerDb, error) {
+// Returns an IndexerDb object, an availability channel that closes when the database
+// becomes available, and an error object.
+func IndexerDbByName(name, arg string, opts IndexerDbOptions, log *log.Logger) (IndexerDb, chan struct{}, error) {
 	if val, ok := indexerFactories[name]; ok {
 		return val.Build(arg, opts, log)
 	}
-	return nil, fmt.Errorf("no IndexerDb factory for %s", name)
+	return nil, nil, fmt.Errorf("no IndexerDb factory for %s", name)
 }
 
 func init() {
diff --git a/idb/migration/migration.go b/idb/migration/migration.go
index 6497569..42d11d4 100644
--- a/idb/migration/migration.go
+++ b/idb/migration/migration.go
@@ -186,22 +186,23 @@ func (m *Migration) update(err error, status string, running bool, blocking bool
 	}
 }
 
-// RunMigrations runs all tasks which have been loaded into the migration. It will update the status accordingly as the
-// migration runs. This call will block execution until it completes and should be run in a go routine if that is not
-// expected.
-func (m *Migration) RunMigrations() {
+// This function always blocks. Closes `ch` when blocking migrations finish
+// running successfully.
+func (m *Migration) runMigrations(ch chan struct{}) {
 	m.log.Printf("Running %d migrations.", len(m.tasks))
+
 	blocking := true
 	for _, task := range m.tasks {
-		if task.MigrationID > m.blockUntil {
+		if blocking && (task.MigrationID > m.blockUntil) {
 			blocking = false
+			close(ch)
 		}
 
 		m.update(nil, StatusActivePrefix+task.Description, true, blocking, task.MigrationID)
-		err := task.Handler()
 
+		err := task.Handler()
 		if err != nil {
-			err := fmt.Errorf("%s%d (%s): %v", StatusErrorPrefix, task.MigrationID, task.Description, err)
+			err := fmt.Errorf("%s%d (%s): %w", StatusErrorPrefix, task.MigrationID, task.Description, err)
 			m.log.WithError(err).Errorf("Migration failed")
 			// If a migration failed, mark that the migration is blocking and terminate.
 			blocking = true
@@ -211,6 +212,19 @@ func (m *Migration) RunMigrations() {
 	}
 
 	m.update(nil, StatusComplete, false, false, -1)
+	if blocking {
+		close(ch)
+	}
 	m.log.Println("Migration finished successfully.")
 	return
 }
+
+// RunMigrations runs all tasks which have been loaded into the migration.
+// It will update the status accordingly as the migration runs.
+// RunMigrations immediately returns a channel which gets closed as soon as the last
+// blocking migration finishes running.
+func (m *Migration) RunMigrations() chan struct{} {
+	res := make(chan struct{})
+	go m.runMigrations(res)
+	return res
+}
diff --git a/idb/postgres/postgres.go b/idb/postgres/postgres.go
index ca837dc..2ce4584 100644
--- a/idb/postgres/postgres.go
+++ b/idb/postgres/postgres.go
@@ -49,11 +49,13 @@ var serializable = sql.TxOptions{Isolation: sql.LevelSerializable} // be a real
 var readonlyRepeatableRead = sql.TxOptions{Isolation: sql.LevelRepeatableRead, ReadOnly: true}
 
 // OpenPostgres is available for creating test instances of postgres.IndexerDb
-func OpenPostgres(connection string, opts idb.IndexerDbOptions, log *log.Logger) (pdb *IndexerDb, err error) {
+// Returns an error object and a channel that gets closed when blocking migrations
+// finish running successfully.
+func OpenPostgres(connection string, opts idb.IndexerDbOptions, log *log.Logger) (*IndexerDb, chan struct{}, error) {
 	db, err := sql.Open("postgres", connection)
 
 	if err != nil {
-		return nil, fmt.Errorf("connecting to postgres: %v", err)
+		return nil, nil, fmt.Errorf("connecting to postgres: %v", err)
 	}
 
 	if strings.Contains(connection, "readonly") {
@@ -64,28 +66,41 @@ func OpenPostgres(connection string, opts idb.IndexerDbOptions, log *log.Logger)
 }
 
 // Allow tests to inject a DB
-func openPostgres(db *sql.DB, opts idb.IndexerDbOptions, logger *log.Logger) (pdb *IndexerDb, err error) {
-	pdb = &IndexerDb{
+func openPostgres(db *sql.DB, opts idb.IndexerDbOptions, logger *log.Logger) (*IndexerDb, chan struct{}, error) {
+	idb := &IndexerDb{
 		readonly: opts.ReadOnly,
 		log:      logger,
 		db:       db,
 	}
 
-	if pdb.log == nil {
-		pdb.log = log.New()
-		pdb.log.SetFormatter(&log.JSONFormatter{})
-		pdb.log.SetOutput(os.Stdout)
-		pdb.log.SetLevel(log.TraceLevel)
+	if idb.log == nil {
+		idb.log = log.New()
+		idb.log.SetFormatter(&log.JSONFormatter{})
+		idb.log.SetOutput(os.Stdout)
+		idb.log.SetLevel(log.TraceLevel)
 	}
 
+	var ch chan struct{}
 	// e.g. a user named "readonly" is in the connection string
-	if !opts.ReadOnly {
-		err = pdb.init(opts)
+	if opts.ReadOnly {
+		migrationState, err := idb.getMigrationState()
 		if err != nil {
-			return nil, fmt.Errorf("initializing postgres: %v", err)
+			return nil, nil, fmt.Errorf("openPostgres() err: %w", err)
+		}
+
+		ch = make(chan struct{})
+		if !migrationStateBlocked(migrationState) {
+			close(ch)
+		}
+	} else {
+		var err error
+		ch, err = idb.init(opts)
+		if err != nil {
+			return nil, nil, fmt.Errorf("initializing postgres: %v", err)
 		}
 	}
-	return
+
+	return idb, ch, nil
 }
 
 // IndexerDb is an idb.IndexerDB implementation
@@ -149,25 +164,29 @@ func (db *IndexerDb) isSetup() (bool, error) {
 	return true, nil
 }
 
-func (db *IndexerDb) init(opts idb.IndexerDbOptions) error {
+// Returns an error object and a channel that gets closed when blocking migrations
+// finish running successfully.
+func (db *IndexerDb) init(opts idb.IndexerDbOptions) (chan struct{}, error) {
 	setup, err := db.isSetup()
 	if err != nil {
-		return fmt.Errorf("init() err: %w", err)
+		return nil, fmt.Errorf("init() err: %w", err)
 	}
 
 	if !setup {
 		// new database, run setup
 		_, err = db.db.Exec(setup_postgres_sql)
 		if err != nil {
-			return fmt.Errorf("unable to setup postgres: %v", err)
+			return nil, fmt.Errorf("unable to setup postgres: %v", err)
 		}
 
 		err = db.markMigrationsAsDone()
 		if err != nil {
-			return fmt.Errorf("unable to confirm migration: %v", err)
+			return nil, fmt.Errorf("unable to confirm migration: %v", err)
 		}
 
-		return nil
+		ch := make(chan struct{})
+		close(ch)
+		return ch, nil
 	}
 
 	// see postgres_migrations.go
diff --git a/idb/postgres/postgres_factory.go b/idb/postgres/postgres_factory.go
index 59f950a..1d3c629 100644
--- a/idb/postgres/postgres_factory.go
+++ b/idb/postgres/postgres_factory.go
@@ -13,7 +13,7 @@ func (df postgresFactory) Name() string {
 	return "postgres"
 }
 
-func (df postgresFactory) Build(arg string, opts idb.IndexerDbOptions, log *log.Logger) (idb.IndexerDb, error) {
+func (df postgresFactory) Build(arg string, opts idb.IndexerDbOptions, log *log.Logger) (idb.IndexerDb, chan struct{}, error) {
 	return OpenPostgres(arg, opts, log)
 }
 
diff --git a/idb/postgres/postgres_migrations.go b/idb/postgres/postgres_migrations.go
index 8359a32..4afb78d 100644
--- a/idb/postgres/postgres_migrations.go
+++ b/idb/postgres/postgres_migrations.go
@@ -102,12 +102,14 @@ func upsertMigrationState(db *IndexerDb, tx *sql.Tx, state *MigrationState) erro
 	return db.setMetastate(tx, migrationMetastateKey, string(migrationStateJSON))
 }
 
-func (db *IndexerDb) runAvailableMigrations() (err error) {
+// Returns an error object and a channel that gets closed when blocking migrations
+// finish running successfully.
+func (db *IndexerDb) runAvailableMigrations() (chan struct{}, error) {
 	state, err := db.getMigrationState()
 	if err == idb.ErrorNotInitialized {
 		state = MigrationState{}
 	} else if err != nil {
-		return fmt.Errorf("runAvailableMigrations() err: %w", err)
+		return nil, fmt.Errorf("runAvailableMigrations() err: %w", err)
 	}
 
 	// Make migration tasks
@@ -136,12 +138,11 @@ func (db *IndexerDb) runAvailableMigrations() (err error) {
 
 	db.migration, err = migration.MakeMigration(tasks, db.log)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	go db.migration.RunMigrations()
-
-	return nil
+	ch := db.migration.RunMigrations()
+	return ch, nil
 }
 
 // after setting up a new database, mark state as if all migrations had been done
diff --git a/importer/helper.go b/importer/helper.go
index b2fc11f..84e6b31 100644
--- a/importer/helper.go
+++ b/importer/helper.go
@@ -266,13 +266,19 @@ func updateAccounting(db idb.IndexerDb, frozenCache map[uint64]bool, filter idb.
 	lastRoundsSeen := roundsSeen
 	txnForRound := 0
 	var blockHeaderPtr *types.BlockHeader = nil
+
+	commit := func() {
+		// Don't commit if there were no transactions.
+		if blockHeaderPtr != nil && txnForRound > 0 {
+			err := db.CommitRoundAccounting(act.RoundUpdates, currentRound, blockHeaderPtr)
+			maybeFail(err, l, "failed to commit round accounting")
+		}
+	}
+
 	for txn := range txns {
 		maybeFail(txn.Error, l, "updateAccounting txn fetch, %v", txn.Error)
 		if txn.Round != currentRound {
-			if blockHeaderPtr != nil && txnForRound > 0 {
-				err := db.CommitRoundAccounting(act.RoundUpdates, currentRound, blockHeaderPtr)
-				maybeFail(err, l, "failed to commit round accounting")
-			}
+			commit()
 
 			// initialize accounting for next round
 			txnForRound = 0
@@ -311,10 +317,7 @@ func updateAccounting(db idb.IndexerDb, frozenCache map[uint64]bool, filter idb.
 	}
 
 	// Commit the final round
-	if blockHeaderPtr != nil && txnForRound > 0 {
-		err := db.CommitRoundAccounting(act.RoundUpdates, currentRound, blockHeaderPtr)
-		maybeFail(err, l, "failed to commit round accounting")
-	}
+	commit()
 
 	rounds += roundsSeen
 	if rounds > 0 {
diff --git a/mule/sign.sh b/mule/sign.sh
index 87ec0f0..c2fb98a 100755
--- a/mule/sign.sh
+++ b/mule/sign.sh
@@ -7,8 +7,8 @@ echo
 date "+build_indexer begin SIGN stage %Y%m%d_%H%M%S"
 echo
 
-ARCH=$(./mule/scripts/archtype.sh)
-OS_TYPE=$(./mule/scripts/ostype.sh)
+ARCH=${ARCH:-(./mule/scripts/archtype.sh)}
+OS_TYPE=${OS_TYPE:-$(./mule/scripts/ostype.sh)}
 VERSION=$(./mule/scripts/compute_build_number.sh)
 PKG_DIR="./tmp/node_pkgs/$OS_TYPE/$ARCH/$VERSION"
 SIGNING_KEY_ADDR=dev@algorand.com
diff --git a/util/metrics/metrics.go b/util/metrics/metrics.go
new file mode 100644
index 0000000..f68cee6
--- /dev/null
+++ b/util/metrics/metrics.go
@@ -0,0 +1,59 @@
+package metrics
+
+import "github.com/prometheus/client_golang/prometheus"
+
+// RegisterPrometheusMetrics register all prometheus metrics with the global
+// metrics handler.
+func RegisterPrometheusMetrics() {
+	prometheus.Register(BlockImportTimeSeconds)
+	prometheus.Register(ImportedTxnsPerBlock)
+	prometheus.Register(ImportedRoundGauge)
+	prometheus.Register(BlockUploadTimeSeconds)
+}
+
+// Prometheus metric names broken out for reuse.
+const (
+	BlockImportTimeName      = "import_time_sec"
+	BlockUploadTimeName      = "block_upload_time_sec"
+	ImportedTxnsPerBlockName = "imported_tx_per_block"
+	ImportedRoundGaugeName   = "imported_round"
+)
+
+// AllMetricNames is a reference for all the custom metric names.
+var AllMetricNames = []string{
+	BlockImportTimeName,
+	BlockUploadTimeName,
+	ImportedTxnsPerBlockName,
+	ImportedRoundGaugeName,
+}
+
+// Initialize the prometheus objects.
+var (
+	BlockImportTimeSeconds = prometheus.NewSummary(
+		prometheus.SummaryOpts{
+			Subsystem: "indexer_daemon",
+			Name:      BlockImportTimeName,
+			Help:      "Total block upload and processing time in seconds.",
+		})
+
+	BlockUploadTimeSeconds = prometheus.NewSummary(
+		prometheus.SummaryOpts{
+			Subsystem: "indexer_daemon",
+			Name:      BlockUploadTimeName,
+			Help:      "Block upload time in seconds.",
+		})
+
+	ImportedTxnsPerBlock = prometheus.NewSummary(
+		prometheus.SummaryOpts{
+			Subsystem: "indexer_daemon",
+			Name:      ImportedTxnsPerBlockName,
+			Help:      "Transactions per block.",
+		})
+
+	ImportedRoundGauge = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Subsystem: "indexer_daemon",
+			Name:      ImportedRoundGaugeName,
+			Help:      "The most recent round indexer has imported.",
+		})
+)


Test patch:
diff --git a/cmd/e2equeries/main.go b/cmd/e2equeries/main.go
index 581b4ee..b57b195 100644
--- a/cmd/e2equeries/main.go
+++ b/cmd/e2equeries/main.go
@@ -34,8 +34,9 @@ func main() {
 	flag.Parse()
 	testutil.SetQuiet(quiet)
 
-	db, err := idb.IndexerDbByName("postgres", pgdb, idb.IndexerDbOptions{ReadOnly: true}, nil)
+	db, availableCh, err := idb.IndexerDbByName("postgres", pgdb, idb.IndexerDbOptions{ReadOnly: true}, nil)
 	maybeFail(err, "open postgres, %v", err)
+	<-availableCh
 
 	rekeyTxnQuery := idb.TransactionFilter{RekeyTo: &truev, Limit: 1}
 	printTxnQuery(db, rekeyTxnQuery)
diff --git a/cmd/idbtest/idbtest.go b/cmd/idbtest/idbtest.go
index e7267f0..f02909f 100644
--- a/cmd/idbtest/idbtest.go
+++ b/cmd/idbtest/idbtest.go
@@ -126,8 +126,10 @@ func main() {
 	flag.Parse()
 	testutil.SetQuiet(quiet)
 
-	db, err := idb.IndexerDbByName("postgres", pgdb, idb.IndexerDbOptions{}, nil)
+	db, availableCh, err :=
+		idb.IndexerDbByName("postgres", pgdb, idb.IndexerDbOptions{}, nil)
 	maybeFail(err, "open postgres, %v", err)
+	<-availableCh
 
 	if accounttest {
 		printAccountQuery(db, idb.AccountQueryOptions{IncludeAssetHoldings: true, IncludeAssetParams: true, AlgosGreaterThan: uint64Ptr(10000000000), Limit: 20})
diff --git a/idb/migration/migration_test.go b/idb/migration/migration_test.go
index eb24a6d..03d9919 100644
--- a/idb/migration/migration_test.go
+++ b/idb/migration/migration_test.go
@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -288,7 +289,7 @@ func TestSuccessfulMigration(t *testing.T) {
 			start := time.Now()
 			migChan := make(chan struct{})
 			go func() {
-				migration.RunMigrations()
+				migration.runMigrations(make(chan struct{}))
 				migChan <- struct{}{}
 			}()
 
@@ -329,3 +330,117 @@ func TestSuccessfulMigration(t *testing.T) {
 		})
 	}
 }
+
+// TestAvailabilityChannelCloses tests that the migration object closes the availability
+// channel when blocking migrations finish.
+func TestAvailabilityChannelCloses(t *testing.T) {
+	// Migration 2 reads on this channel.
+	migrationTwoChannel := make(chan struct{})
+	defer func() {
+		migrationTwoChannel <- struct{}{}
+	}()
+
+	tasks := []Task{
+		{
+			MigrationID: 1,
+			Handler: func() error {
+				return nil
+			},
+			DBUnavailable: true,
+		},
+		{
+			MigrationID: 2,
+			Handler: func() error {
+				<-migrationTwoChannel
+				return nil
+			},
+		},
+	}
+
+	m, err := MakeMigration(tasks, nil)
+	require.NoError(t, err)
+
+	availableCh := m.RunMigrations()
+	select {
+	case _, ok := <-availableCh:
+		assert.False(t, ok)
+	case <-time.After(10 * time.Millisecond):
+		assert.Fail(t, "channel must be closed")
+	}
+}
+
+// TestAvailabilityChannelClosesNoMigrations tests that the migration object closes
+// the availability channel when last migration, which is blocking, finishes.
+func TestAvailabilityChannelClosesNoMigrations(t *testing.T) {
+	tasks := []Task{
+		{
+			MigrationID: 1,
+			Handler: func() error {
+				return nil
+			},
+			DBUnavailable: true,
+		},
+	}
+
+	m, err := MakeMigration(tasks, nil)
+	require.NoError(t, err)
+
+	availableCh := m.RunMigrations()
+	select {
+	case _, ok := <-availableCh:
+		assert.False(t, ok)
+	case <-time.After(10 * time.Millisecond):
+		assert.Fail(t, "channel must be closed")
+	}
+}
+
+// TestAvailabilityChannelClosesBlockingMigrationLast tests that the migration object
+// closes when there are no migrations.
+func TestAvailabilityChannelClosesBlockingMigrationLast(t *testing.T) {
+	m, err := MakeMigration([]Task{}, nil)
+	require.NoError(t, err)
+
+	availableCh := m.RunMigrations()
+	select {
+	case _, ok := <-availableCh:
+		assert.False(t, ok)
+	case <-time.After(10 * time.Millisecond):
+		assert.Fail(t, "channel must be closed")
+	}
+}
+
+// TestAvailabilityChannelDoesNotCloseEarly tests that the migration object closes the availability
+// channel only after all blocking migrations are run.
+func TestAvailabilityChannelDoesNotCloseEarly(t *testing.T) {
+	migrationTwoChannel := make(chan struct{})
+	defer func() {
+		migrationTwoChannel <- struct{}{}
+	}()
+
+	tasks := []Task{
+		{
+			MigrationID: 1,
+			Handler: func() error {
+				return nil
+			},
+		},
+		{
+			MigrationID: 2,
+			Handler: func() error {
+				<-migrationTwoChannel
+				return nil
+			},
+			DBUnavailable: true,
+		},
+	}
+
+	m, err := MakeMigration(tasks, nil)
+	require.NoError(t, err)
+
+	availableCh := m.RunMigrations()
+	select {
+	case <-availableCh:
+		assert.Fail(t, "availability channel closed before migrations finish running")
+	case <-time.After(5 * time.Millisecond):
+	}
+}
diff --git a/idb/postgres/postgres_integration_common_test.go b/idb/postgres/postgres_integration_common_test.go
index ce1a1aa..24670fa 100644
--- a/idb/postgres/postgres_integration_common_test.go
+++ b/idb/postgres/postgres_integration_common_test.go
@@ -2,6 +2,7 @@ package postgres
 
 import (
 	"database/sql"
+	"flag"
 	"fmt"
 	"testing"
 
@@ -17,8 +18,24 @@ import (
 	"github.com/algorand/indexer/util/test"
 )
 
+var testpg = flag.String("test-pg", "", "postgres connection string")
+
 // setupPostgres starts a gnomock postgres DB then returns the connection string and a shutdown function.
 func setupPostgres(t *testing.T) (*sql.DB, string, func()) {
+	if testpg != nil && *testpg != "" {
+		// TODO: Drop schema?
+
+		// use non-docker Postgresql
+		shutdownFunc := func() {
+			// nothing to do, psql db setup/teardown is external
+		}
+		connStr := *testpg
+		db, err := sql.Open("postgres", connStr)
+		require.NoError(t, err, "Error opening pg connection")
+
+		return db, connStr, shutdownFunc
+	}
+
 	p := postgres.Preset(
 		postgres.WithVersion("12.5"),
 		postgres.WithUser("gnomock", "gnomick"),
@@ -47,7 +64,7 @@ func setupPostgres(t *testing.T) (*sql.DB, string, func()) {
 func setupIdb(t *testing.T, genesis types.Genesis) (*IndexerDb /*db*/, func() /*shutdownFunc*/) {
 	_, connStr, shutdownFunc := setupPostgres(t)
 
-	idb, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
+	idb, _, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
 	require.NoError(t, err)
 
 	err = idb.LoadGenesis(genesis)
diff --git a/idb/postgres/postgres_integration_test.go b/idb/postgres/postgres_integration_test.go
index 1737b8e..e17c692 100644
--- a/idb/postgres/postgres_integration_test.go
+++ b/idb/postgres/postgres_integration_test.go
@@ -36,7 +36,7 @@ func TestMaxRoundOnUninitializedDB(t *testing.T) {
 	_, connStr, shutdownFunc := setupPostgres(t)
 	defer shutdownFunc()
 
-	db, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
+	db, _, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
 	assert.NoError(t, err)
 
 	round, err := db.GetNextRoundToAccount()
@@ -57,7 +57,7 @@ func TestMaxRoundEmptyMetastate(t *testing.T) {
 	pg, connStr, shutdownFunc := setupPostgres(t)
 	defer shutdownFunc()
 
-	db, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
+	db, _, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
 	assert.NoError(t, err)
 	pg.Exec(`INSERT INTO metastate (k, v) values ('state', '{}')`)
 
@@ -75,7 +75,7 @@ func TestMaxRound(t *testing.T) {
 	db, connStr, shutdownFunc := setupPostgres(t)
 	defer shutdownFunc()
 
-	pdb, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
+	pdb, _, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
 	assert.NoError(t, err)
 	db.Exec(
 		`INSERT INTO metastate (k, v) values ($1, $2)`,
@@ -103,7 +103,7 @@ func TestAccountedRoundNextRound0(t *testing.T) {
 	db, connStr, shutdownFunc := setupPostgres(t)
 	defer shutdownFunc()
 
-	pdb, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
+	pdb, _, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
 	assert.NoError(t, err)
 	db.Exec(
 		`INSERT INTO metastate (k, v) values ($1, $2)`,
@@ -226,7 +226,7 @@ func TestInitializeFrozenCache(t *testing.T) {
 	defer shutdownFunc()
 
 	// Initialize DB by creating one of these things.
-	_, err := idb.IndexerDbByName("postgres", connStr, idb.IndexerDbOptions{}, nil)
+	_, _, err := idb.IndexerDbByName("postgres", connStr, idb.IndexerDbOptions{}, nil)
 	assert.NoError(t, err)
 
 	// Add some assets
@@ -243,7 +243,7 @@ func TestInitializeFrozenCache(t *testing.T) {
 		3, test.AccountA[:], `{}`)
 	require.NoError(t, err)
 
-	pdb, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
+	pdb, _, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
 	assert.NoError(t, err)
 	cache, err := pdb.GetDefaultFrozen()
 	assert.NoError(t, err)
@@ -414,7 +414,7 @@ func TestBlockWithTransactions(t *testing.T) {
 	db, connStr, shutdownFunc := setupPostgres(t)
 	defer shutdownFunc()
 
-	pdb, err := idb.IndexerDbByName("postgres", connStr, idb.IndexerDbOptions{}, nil)
+	pdb, _, err := idb.IndexerDbByName("postgres", connStr, idb.IndexerDbOptions{}, nil)
 	assert.NoError(t, err)
 
 	assetid := uint64(2222)
@@ -1078,14 +1078,18 @@ func TestLargeAssetAmount(t *testing.T) {
 	}
 }
 
-// Test that initializing a new database sets the correct migration number.
+// Test that initializing a new database sets the correct migration number and
+// that the database is available.
 func TestInitializationNewDatabase(t *testing.T) {
 	_, connStr, shutdownFunc := setupPostgres(t)
 	defer shutdownFunc()
 
-	db, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
+	db, availableCh, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
 	require.NoError(t, err)
 
+	_, ok := <-availableCh
+	assert.False(t, ok)
+
 	state, err := db.getMigrationState()
 	require.NoError(t, err)
 
@@ -1097,10 +1101,10 @@ func TestOpenDbAgain(t *testing.T) {
 	_, connStr, shutdownFunc := setupPostgres(t)
 	defer shutdownFunc()
 
-	_, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
+	_, _, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
 	require.NoError(t, err)
 
-	_, err = OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
+	_, _, err = OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
 	require.NoError(t, err)
 }
 
diff --git a/idb/postgres/postgres_migrations_integration_test.go b/idb/postgres/postgres_migrations_integration_test.go
index 90e42ed..323bef7 100644
--- a/idb/postgres/postgres_migrations_integration_test.go
+++ b/idb/postgres/postgres_migrations_integration_test.go
@@ -32,7 +32,7 @@ type oldImportState struct {
 func TestMaxRoundAccountedMigrationAccountRound0(t *testing.T) {
 	_, connStr, shutdownFunc := setupPostgres(t)
 	defer shutdownFunc()
-	db, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
+	db, _, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
 	assert.NoError(t, err)
 
 	round := int64(0)
@@ -64,7 +64,7 @@ func TestMaxRoundAccountedMigrationAccountRound0(t *testing.T) {
 func TestMaxRoundAccountedMigrationAccountRoundPositive(t *testing.T) {
 	_, connStr, shutdownFunc := setupPostgres(t)
 	defer shutdownFunc()
-	db, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
+	db, _, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
 	assert.NoError(t, err)
 
 	round := int64(2)
@@ -96,7 +96,7 @@ func TestMaxRoundAccountedMigrationAccountRoundPositive(t *testing.T) {
 func TestMaxRoundAccountedMigrationUninitialized(t *testing.T) {
 	_, connStr, shutdownFunc := setupPostgres(t)
 	defer shutdownFunc()
-	db, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
+	db, _, err := OpenPostgres(connStr, idb.IndexerDbOptions{}, nil)
 	assert.NoError(t, err)
 
 	migrationState := MigrationState{NextMigration: 4}
diff --git a/idb/postgres/postgres_test.go b/idb/postgres/postgres_test.go
index 834d180..ab4074d 100644
--- a/idb/postgres/postgres_test.go
+++ b/idb/postgres/postgres_test.go
@@ -31,7 +31,7 @@ func TestAllMigrations(t *testing.T) {
 			})
 
 			// This automatically runs migrations
-			pdb, err := openPostgres(db, idb.IndexerDbOptions{
+			pdb, _, err := openPostgres(db, idb.IndexerDbOptions{
 				ReadOnly: false,
 			}, nil)
 			require.NoError(t, err)
@@ -72,7 +72,7 @@ func TestNoMigrationsNeeded(t *testing.T) {
 	})
 
 	// This automatically runs migraions
-	pdb, err := openPostgres(db, idb.IndexerDbOptions{
+	pdb, _, err := openPostgres(db, idb.IndexerDbOptions{
 		ReadOnly: false,
 	}, nil)
 	assert.NoError(t, err)