Merge pull request #2584 from filecoin-project/fix/chainwatch/top-miners-by-reward-applies-wincount

Top Miner by Reward applies WinCount; Add max_height view

cmd/lotus-chainwatch/processor/reward.go

@@ -51,30 +51,11 @@ create table if not exists chain_power
 			primary key,
 	baseline_power text not null
 );
-
-create materialized view if not exists top_miners_by_base_reward as
-	with total_rewards_by_miner as (
-		select
-			b.miner,
-			sum(bbr.base_block_reward) as total_reward
-		from blocks b
-		inner join base_block_rewards bbr on b.parentstateroot = bbr.state_root
-		group by 1
-	) select
-		rank() over (order by total_reward desc),
-		miner,
-		total_reward
-	from total_rewards_by_miner
-	group by 2, 3;
-
-create index if not exists top_miners_by_base_reward_miner_index
-	on top_miners_by_base_reward (miner);
 `); err != nil {
 		return err
 	}
 
 	return tx.Commit()
-
 }
 
 func (p *Processor) HandleRewardChanges(ctx context.Context, rewardTips ActorTips) error {

cmd/lotus-chainwatch/scheduler/refresh_top_miners_by_base_reward.go

@@ -8,6 +8,56 @@ import (
 	"golang.org/x/xerrors"
 )
 
+func setupTopMinerByBaseRewardSchema(ctx context.Context, db *sql.DB) error {
+	select {
+	case <-ctx.Done():
+		return nil
+	default:
+	}
+
+	tx, err := db.Begin()
+	if err != nil {
+		return err
+	}
+	if _, err := tx.Exec(`
+		create materialized view if not exists top_miners_by_base_reward as
+			with total_rewards_by_miner as (
+				select
+					b.miner,
+					sum(bbr.base_block_reward * b.win_count) as total_reward
+				from blocks b
+				inner join base_block_rewards bbr on b.parentstateroot = bbr.state_root
+				group by 1
+			) select
+				rank() over (order by total_reward desc),
+				miner,
+				total_reward
+			from total_rewards_by_miner
+			group by 2, 3;
+
+		create index if not exists top_miners_by_base_reward_miner_index
+			on top_miners_by_base_reward (miner);
+
+		create materialized view if not exists top_miners_by_base_reward_max_height as
+			select
+				b."timestamp"as current_timestamp,
+				max(b.height) as current_height
+			from blocks b
+			join base_block_rewards bbr on b.parentstateroot = bbr.state_root
+			where bbr.base_block_reward is not null
+			group by 1
+			order by 1 desc
+			limit 1;
+	`); err != nil {
+		return xerrors.Errorf("create top_miner_by_base_reward views", err)
+	}
+
+	if err := tx.Commit(); err != nil {
+		return xerrors.Errorf("commiting top_miner_by_base_reward views", err)
+	}
+	return nil
+}
+
 func refreshTopMinerByBaseReward(ctx context.Context, db *sql.DB) error {
 	select {
 	case <-ctx.Done():
@@ -20,10 +70,15 @@ func refreshTopMinerByBaseReward(ctx context.Context, db *sql.DB) error {
 		log.Debugw("refresh top_miners_by_base_reward", "duration", time.Since(t).String())
 	}()
 
-	_, err := db.Exec("REFRESH MATERIALIZED VIEW top_miners_by_base_reward;")
+	_, err := db.Exec("refresh materialized view top_miners_by_base_reward;")
 	if err != nil {
 		return xerrors.Errorf("refresh top_miners_by_base_reward: %w", err)
 	}
 
+	_, err = db.Exec("refresh materialized view top_miners_by_base_reward_max_height;")
+	if err != nil {
+		return xerrors.Errorf("refresh top_miners_by_base_reward_max_height: %w", err)
+	}
+
 	return nil
 }

cmd/lotus-chainwatch/scheduler/scheduler.go

@@ -6,6 +6,8 @@ import (
 	"time"
 
 	logging "github.com/ipfs/go-log/v2"
+
+	"golang.org/x/xerrors"
 )
 
 var log = logging.Logger("scheduler")
@@ -21,10 +23,21 @@ func PrepareScheduler(db *sql.DB) *Scheduler {
 	return &Scheduler{db}
 }
 
+func (s *Scheduler) setupSchema(ctx context.Context) error {
+	if err := setupTopMinerByBaseRewardSchema(ctx, s.db); err != nil {
+		return xerrors.Errorf("setup top miners by reward schema", err)
+	}
+	return nil
+}
+
 // Start the scheduler jobs at the defined intervals
 func (s *Scheduler) Start(ctx context.Context) {
 	log.Debug("Starting Scheduler")
 
+	if err := s.setupSchema(ctx); err != nil {
+		log.Fatalw("applying scheduling schema", err)
+	}
+
 	go func() {
 		// run once on start after schema has initialized
 		time.Sleep(5 * time.Second)