Repository: algorand/indexer
Pull request #586 (2021-07-22 20:52:29)

daemon: replace a sleep loop with a channel.
## Summary

Currently, `daemon.go` has a sleep loop calls `idb.Health()` until it reports that the database is available (blocking migrations have run successfully). A sleep loop is not elegant, and `idb.Health()` might return an error because it can't parse data from the database before blocking migrations have run. Particularly now it is possible that `idb.Health()` is run before the blocking migration that changes the import state format, in which case indexer will crash.

This PR replaces the sleep loop with a channel that is closed by the migration object as soon as all blocking migrations have finished running successfully.

## Test Plan

Added tests for the `Migration` type.

Tested manually that
- Indexer runs normally with an empty database
- Indexer runs normally with a database that already has some blocks (from previous scenario)
- When upgrading from 2.5.0, Indexer starts importing blocks only after the blocking migration finishes.

codecov-commenter: # [Codecov](https://codecov.io/gh/algorand/indexer/pull/586?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) Report
> Merging [#586](https://codecov.io/gh/algorand/indexer/pull/586?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (3e24c1b) 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 **increase** coverage by `0.06%`.
> The diff coverage is `65.00%`.

[![Impacted file tree graph](https://codecov.io/gh/algorand/indexer/pull/586/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/586?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)

```diff
@@             Coverage Diff             @@
##           develop     #586      +/-   ##
===========================================
+ Coverage    47.50%   47.56%   +0.06%     
===========================================
  Files           25       25              
  Lines         3846     3864      +18     
===========================================
+ Hits          1827     1838      +11     
- Misses        1739     1746       +7     
  Partials       280      280              
```


| [Impacted Files](https://codecov.io/gh/algorand/indexer/pull/586?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [idb/postgres/postgres\_factory.go](https://codecov.io/gh/algorand/indexer/pull/586/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-aWRiL3Bvc3RncmVzL3Bvc3RncmVzX2ZhY3RvcnkuZ28=) | `66.66% <ø> (ø)` | |
| [idb/postgres/postgres\_migrations.go](https://codecov.io/gh/algorand/indexer/pull/586/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-aWRiL3Bvc3RncmVzL3Bvc3RncmVzX21pZ3JhdGlvbnMuZ28=) | `63.23% <50.00%> (ø)` | |
| [idb/postgres/postgres.go](https://codecov.io/gh/algorand/indexer/pull/586/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-aWRiL3Bvc3RncmVzL3Bvc3RncmVzLmdv) | `43.12% <55.55%> (-0.01%)` | :arrow_down: |
| [idb/migration/migration.go](https://codecov.io/gh/algorand/indexer/pull/586/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-aWRiL21pZ3JhdGlvbi9taWdyYXRpb24uZ28=) | `96.10% <100.00%> (+0.32%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/algorand/indexer/pull/586?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/586?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None). Last update [6e4d737...3e24c1b](https://codecov.io/gh/algorand/indexer/pull/586?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/cmd/algorand-indexer/daemon.go b/cmd/algorand-indexer/daemon.go
index 8fb6e39..93c2feb 100644
--- a/cmd/algorand-indexer/daemon.go
+++ b/cmd/algorand-indexer/daemon.go
@@ -79,30 +79,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 +120,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")
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..645adf3 100644
--- a/cmd/algorand-indexer/main.go
+++ b/cmd/algorand-indexer/main.go
@@ -84,18 +84,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
 	}
-	return db
+	if dummyIndexerDb {
+		return dummy.IndexerDb(), nil
+	}
+	logger.Errorf("no import db set")
+	os.Exit(1)
+	return nil, nil
 }
 
 func init() {
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


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..d99fe8e 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,112 @@ 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) {
+	tasks := []Task{
+		{
+			MigrationID: 1,
+			Handler: func() error {
+				return nil
+			},
+		},
+		{
+			MigrationID: 2,
+			Handler: func() error {
+				time.Sleep(10 * time.Millisecond)
+				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..fe24b15 100644
--- a/idb/postgres/postgres_integration_common_test.go
+++ b/idb/postgres/postgres_integration_common_test.go
@@ -47,7 +47,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)