add pyth_aggregated_slot metric

Author: Richard Patel

main.go

@@ -105,7 +105,7 @@ func main() {
 	log.Info("Listing all products")
 	productCache, products, err := discoverProducts(initContext, client)
 	if err != nil {
-		log.Fatal("Failed to list products")
+		log.Fatal("Failed to list products", zap.Error(err))
 	}
 	if len(allPublishers) == 0 {
 		log.Info("Listing all publishers")
@@ -193,7 +193,7 @@ func main() {
 
 	if len(allPublishers) > 0 {
 		// Scrape publisher balances.
-		balances := newBalanceScraper(publishKeys.pubkeys, *rpcURL, log.Named("balances"))
+		balances := newBalanceScraper(allPublishers, *rpcURL, log.Named("balances"))
 		metrics.Registry.MustRegister(balances)
 
 		// Create tx tailer.

metrics/metrics.go

@@ -71,6 +71,12 @@ var (
 		Name:      "aggregated_conf_amount",
 		Help:      "Last aggregated conf of Pyth product",
 	}, []string{labelProduct, labelSymbol})
+	AggSlot = factory.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: Namespace,
+		Subsystem: SubsystemOracle,
+		Name:      "aggregated_slot",
+		Help:      "Last observed slot for Pyth product",
+	}, []string{labelProduct, labelSymbol})
 	AggStatus = factory.NewGaugeVec(prometheus.GaugeOpts{
 		Namespace: Namespace,
 		Subsystem: SubsystemOracle,

price.go

@@ -70,6 +70,9 @@ func (p *priceScraper) aggregate(product *solana.PublicKey, agg *pyth.PriceInfo,
 	metrics.AggConf.
 		WithLabelValues(productStr, symbol).
 		Set(float64(agg.Conf) * decimals)
+	metrics.AggSlot.
+		WithLabelValues(productStr, symbol).
+		Set(float64(agg.PubSlot))
 	metrics.AggStatus.
 		WithLabelValues(productStr, symbol).
 		Set(float64(agg.Status))

txs.go

@@ -83,12 +83,12 @@ func (s *txScraper) pollOne(ctx context.Context, wg *sync.WaitGroup, tailer *txT
 	defer wg.Done()
 	for {
 		isEnd, err := tailer.poll(ctx)
-		if isEnd {
-			break
-		}
 		if err != nil {
 			s.log.Warn("Failed to poll account txs", zap.Error(err))
 		}
+		if isEnd {
+			break
+		}
 	}
 }
 
@@ -111,6 +111,9 @@ func newTxTailer(scraper *txScraper, pubkey solana.PublicKey) *txTailer {
 }
 
 func (t *txTailer) refreshLastSig(ctx context.Context) error {
+	t.log.Debug("Getting latest sig",
+		zap.String("publisher", t.pubkeyStr))
+
 	oneInt := 1
 	sigs, err := t.rpc.GetSignaturesForAddressWithOpts(ctx, t.pubkey, &rpc.GetSignaturesForAddressOpts{
 		Limit: &oneInt,
@@ -121,10 +124,17 @@ func (t *txTailer) refreshLastSig(ctx context.Context) error {
 	metrics.RpcRequestsTotal.Inc()
 
 	if len(sigs) == 0 {
+		t.log.Debug("Publisher has not sent any txs yet",
+			zap.String("publisher", t.pubkeyStr))
 		return nil // empty account
 	}
 	t.lastSlot = sigs[0].Slot
 	t.lastSig = sigs[0].Signature
+
+	t.log.Debug("Tailing txs starting at",
+		zap.String("publisher", t.pubkeyStr),
+		zap.Stringer("start_sig", t.lastSig))
+
 	return nil
 }