core/account: add expiration block processor

This runs the expiration of account control programs as a separate
process. This fixes the expiration to wait until the transaction
processor has completed before deleting any data that may be needed.

1.1-stable version of #612

Closes #679

cmd/cored/main.go

@@ -312,6 +312,7 @@ func launchConfiguredCore(ctx context.Context, db *sql.DB, conf *config.Config,
 	}
 	// Start listeners
 	go pinStore.Listen(ctx, account.PinName, *dbURL)
+	go pinStore.Listen(ctx, account.ExpirePinName, *dbURL)
 	go pinStore.Listen(ctx, asset.PinName, *dbURL)
 
 	// Setup the transaction query indexer to index every transaction.
@@ -392,6 +393,10 @@ func launchConfiguredCore(ctx context.Context, db *sql.DB, conf *config.Config,
 		if err != nil {
 			chainlog.Fatal(ctx, chainlog.KeyError, err)
 		}
+		err = pinStore.CreatePin(ctx, account.ExpirePinName, height)
+		if err != nil {
+			chainlog.Fatal(ctx, chainlog.KeyError, err)
+		}
 		err = pinStore.CreatePin(ctx, asset.PinName, height)
 		if err != nil {
 			chainlog.Fatal(ctx, chainlog.KeyError, err)

core/account/indexer.go

@@ -14,9 +14,14 @@ import (
 	"chain/protocol/bc"
 )
 
-// PinName is used to identify the pin associated with
-// the account block processor.
-const PinName = "account"
+const (
+	// PinName is used to identify the pin associated with
+	// the account indexer block processor.
+	PinName = "account"
+	// ExpirePinName is used to identify the pin associated
+	// with the account control program expiration processor.
+	ExpirePinName = "expire-control-programs"
+)
 
 var emptyJSONObject = json.RawMessage(`{}`)
 
@@ -92,18 +97,17 @@ func (m *Manager) ProcessBlocks(ctx context.Context) {
 	if m.pinStore == nil {
 		return
 	}
-	m.pinStore.ProcessBlocks(ctx, m.chain, PinName, m.processBlock)
+	go m.pinStore.ProcessBlocks(ctx, m.chain, ExpirePinName, m.expireControlPrograms)
+	m.pinStore.ProcessBlocks(ctx, m.chain, PinName, m.indexAccountUTXOs)
 }
 
-func (m *Manager) processBlock(ctx context.Context, b *bc.Block) error {
-	err := m.indexAccountUTXOs(ctx, b)
-	if err != nil {
-		return err
-	}
+func (m *Manager) expireControlPrograms(ctx context.Context, b *bc.Block) error {
+	<-m.pinStore.PinWaiter(PinName, b.Height)
+	<-m.pinStore.PinWaiter(query.TxPinName, b.Height)
 
 	// Delete expired account control programs.
 	const deleteQ = `DELETE FROM account_control_programs WHERE expires_at IS NOT NULL AND expires_at < $1`
-	_, err = m.db.Exec(ctx, deleteQ, b.Time())
+	_, err := m.db.Exec(ctx, deleteQ, b.Time())
 	return err
 }