Merge pull request #2910 from filecoin-project/feat/chainwatch/capture-circulating-fil

feat(chainwatch): capture circulating fil per tipset metric

cmd/lotus-chainwatch/syncer/sync.go

@@ -4,6 +4,7 @@ import (
 	"container/list"
 	"context"
 	"database/sql"
+	"fmt"
 	"sync"
 	"time"
 
@@ -40,6 +41,14 @@ func (s *Syncer) setupSchemas() error {
 	}
 
 	if _, err := tx.Exec(`
+/* tracks circulating fil available on the network at each tipset */
+create table if not exists chain_economics
+(
+	parent_state_root text not null
+		constraint chain_economics_pk primary key,
+	circulating_fil text not null
+);
+
 create table if not exists block_cids
 (
 	cid text not null
@@ -173,6 +182,10 @@ func (s *Syncer) Start(ctx context.Context) {
 						log.Errorw("failed to gather unsynced blocks", "error", err)
 					}
 
+					if err := s.storeCirculatingSupply(ctx, change.Val); err != nil {
+						log.Errorw("failed to store circulating supply", "error", err)
+					}
+
 					if len(unsynced) == 0 {
 						continue
 					}
@@ -264,6 +277,24 @@ func (s *Syncer) syncedBlocks(timestamp time.Time) (map[cid.Cid]struct{}, error)
 	return out, nil
 }
 
+func (s *Syncer) storeCirculatingSupply(ctx context.Context, tipset *types.TipSet) error {
+	supply, err := s.node.StateCirculatingSupply(ctx, tipset.Key())
+	if err != nil {
+		return err
+	}
+
+	ceInsert := `insert into chain_economics (parent_state_root, circulating_fil) values ('%s', '%s');`
+
+	if _, err := s.db.Exec(fmt.Sprintf(ceInsert,
+		tipset.ParentState().String(),
+		supply.String(),
+	)); err != nil {
+		return xerrors.Errorf("insert circulating supply for tipset (%s): %w", tipset.Key().String(), err)
+	}
+
+	return nil
+}
+
 func (s *Syncer) storeHeaders(bhs map[cid.Cid]*types.BlockHeader, sync bool, timestamp time.Time) error {
 	s.headerLk.Lock()
 	defer s.headerLk.Unlock()